Код:
def fetch_first_page():
"""Подбирает размер страницы (page_size) и возвращает (размер, страницу 1).
Пробуем 200/100/50/30: сервер либо отдаёт столько, сколько попросили,
либо молча режет до своего максимума — берём фактически полученное.
"""
for size in (200, 100, 50, 30):
try:
data = fetch_json(f"{LIST_URL}?page=1&page_size={size}")
except NotFound:
continue
except (HTTPError, URLError, TimeoutError, OSError) as error:
print(f" page_size={size}: не вышел ({error}), пробую меньше")
continue
if not isinstance(data, dict):
continue
results = data.get("results")
if isinstance(results, list) and results:
actual = len(results)
if actual < size:
print(f" сервер ограничил выдачу: просили {size}, "
f"получилось {actual}")
return actual, data
raise RuntimeError("API не отдал даже первую страницу списка")
def ordering_is_ascending(page_data):
"""Идут ли номера в списке по возрастанию с единицы.
Если да — умеем вычислять страницу продолжения.
"""
results = (page_data.get("results") or []) if isinstance(page_data, dict) else []
ids = [item.get("id") for item in results if isinstance(item, dict)]
if not ids or not all(isinstance(i, int) for i in ids):
return False
return ids[0] == 1 and all(a < b for a, b in zip(ids, ids[1:]))
def first_missing_id(known_ids):
"""Первый номер, которого нет в базе (номера идут с 1)."""
candidate = 1
while candidate in known_ids:
candidate += 1
return candidate
def find_start_page(min_missing, page_size):
"""Страница списка, с которой продолжать скачивание.
Список идёт по возрастанию id (проверено), поэтому номер страницы
можно оценить формулой и уточнить запросами: из-за дыр в нумерации
(стихи удаляли) оценка уезжает вперёд — сдвигаемся, пока не накроем
min_missing. Возвращает (страница, данные страницы) — страницу не
придётся качать повторно. Особые случаи: ("end", None) — ничего
нового нет; (None, None) — не получилось, качаем с первой страницы.
"""
page = max(1, (min_missing - 1) // page_size + 1)
for _ in range(30):
try:
data = fetch_json(f"{LIST_URL}?page={page}&page_size={page_size}")
except NotFound:
return "end", None
ids = [item.get("id") for item in (data.get("results") or [])
if isinstance(item, dict)]
if not ids:
return None, None
if max(ids) < min_missing:
page += 1 # ещё не дошли до нужного номера
elif min(ids) > min_missing and page > 1:
page -= 1 # перескочили из-за дыр в номерах
else:
return page, data
return None, None
def do_crawl(conn, args, known_ids, legacy, limit):
"""Обход списка стихов через API: новые добавляем, скачанные обновляем."""
print("Подключаюсь к API...")
if args.page_size:
page_size = args.page_size
pending = fetch_json(f"{LIST_URL}?page=1&page_size={page_size}")
print(f"Размер страницы (задан вручную): {page_size}")
else:
page_size, pending = fetch_first_page()
count = pending.get("count") if isinstance(pending, dict) else None
total_pages = None
if isinstance(count, int) and count > 0:
total_pages = (count + page_size - 1) // page_size
print(f"На сайте {count} стихов; страниц списка: {total_pages} "
f"(по {page_size} шт.)")
else:
print(f"Сервер не сообщил общее число стихов; страница: {page_size} шт.")
if not args.trust_ids:
check_id_mapping(conn)
start_page = 1
if args.refresh:
print("Режим --refresh: прохожу список с первой страницы — "
"скачанное обновится, новое добавится")
elif known_ids or args.start > 1:
min_missing = args.start if args.start > 1 else first_missing_id(known_ids)
if ordering_is_ascending(pending):
page, data = find_start_page(min_missing, page_size)
if page == "end":
print("Все стихи до конца списка уже в базе — нового нет")
return 0, 0, 0, 0
if page is not None:
start_page, pending = page, data
print(f"Продолжаю со страницы {start_page} "
f"(первый отсутствующий номер: {min_missing})")
else:
print("Не определил страницу продолжения — качаю с первой, "
"скачанные номера будут обновляться на месте")
else:
print("Список в API не по возрастанию номеров — качаю с первой "
"страницы, скачанные номера будут обновляться на месте")
page = start_page
processed = added = enriched = refreshed = empty = 0
pages_done = 0
started = time.monotonic()
try:
while processed < limit:
if pending is not None:
data, pending = pending, None
else:
data = fetch_json(f"{LIST_URL}?page={page}&page_size={page_size}")
results = data.get("results") or []
if not results:
print(f"Страница {page}: пустая — список кончился")
break
page_added = page_refreshed = 0
for item in results:
if processed >= limit:
break
try:
site_id = int(item["id"])
except (KeyError, TypeError, ValueError):
continue
if args.start > 1 and site_id < args.start:
continue
poem = item_to_poem(item)
if poem is None:
empty += 1
continue
processed += 1
text, author, date, section, likes, rating = poem
key = normalize(text)
if site_id in known_ids:
# Уже в базе: обновляем лайки/рейтинг, дозаполняем остальное
conn.execute(
"UPDATE poems SET likes = COALESCE(?, likes), "
"rating = COALESCE(?, rating), "
"section = COALESCE(section, ?), "
"date = COALESCE(date, ?), "
"author = COALESCE(NULLIF(author, ''), ?) "
"WHERE site_id = ?",
(likes, rating, section, date, author, site_id),
)
known_ids.add(site_id)
refreshed += 1
page_refreshed += 1
elif key in legacy:
# Такой текст уже лежал в базе без номера — дополняем
conn.execute(
"UPDATE poems SET site_id = ?, "
"author = COALESCE(NULLIF(author, ''), ?), "
"date = COALESCE(date, ?), section = COALESCE(section, ?), "
"likes = COALESCE(likes, ?), rating = COALESCE(rating, ?) "
"WHERE rowid = ?",
(site_id, author, date, section, likes, rating,
legacy.pop(key)),
)
known_ids.add(site_id)
enriched += 1
page_added += 1
else:
conn.execute(
"INSERT INTO poems (site_id, content, author, section, "
"date, likes, rating) VALUES (?, ?, ?, ?, ?, ?, ?)",
(site_id, text, author, section, date, likes, rating),
)
known_ids.add(site_id)
added += 1
page_added += 1
conn.commit()
pages_done += 1
line = f"Страница {page:>6}"
if total_pages:
line += f"/{total_pages}"
print(f"{line}: новых {page_added}, обновлено {page_refreshed}")
if pages_done % 10 == 0:
elapsed = time.monotonic() - started
if elapsed > 0:
speed = pages_done / elapsed
eta = f", осталось ~{(total_pages - page) / speed / 60:.0f} мин" \
if total_pages else ""
print(f" … {pages_done} стр., {speed * 60:.0f} стр/мин{eta}")
if not data.get("next"):
print("Поле next пустое — дошли до конца списка")
break
page += 1
# Случайная пауза, чтобы запросы не шли как по линейке
time.sleep(random.uniform(args.delay, args.delay * 2))
except KeyboardInterrupt:
print("\nОстановлено вручную (Ctrl+C) — всё скачанное уже сохранено в базе")
except NotFound:
print("\nСервер ответил 404 — страницы списка кончились (штатный конец)")
except (HTTPError, URLError, TimeoutError, OSError, RuntimeError) as error:
print(f"\nСбой: {error}")
print("Продолжить: py scrape_poetory.py --all "
"(скрипт сам найдёт, где остановились)")
return added, enriched, refreshed, empty
def main():
parser = argparse.ArgumentParser(
description="Скачивает стихи poetory.ru через API в poems.db")
parser.add_argument("count", nargs="?", type=int, default=None,
help="сколько стихов обойти (по умолчанию 100; "
"с --all или --refresh — все)")
parser.add_argument("--all", action="store_true",
help="скачать ВСЁ (по списку API)")
parser.add_argument("--refresh", action="store_true",
help="пройти весь список с начала: обновить лайки/"
"рейтинг у скачанных и добавить новые")
parser.add_argument("--start", type=int, default=1,
help="номер первого стиха (по умолчанию 1; "
"для продолжения не нужен — определяется сам)")
parser.add_argument("--page-size", type=int, default=None,
help="стихов в одном запросе (по умолчанию "
"подбирается сам)")
parser.add_argument("--delay", type=float, default=1.0,
help="минимальная пауза между запросами, сек "
"(по умолчанию 1.0)")
parser.add_argument("--trust-ids", action="store_true",
help="не проверять совпадение id API с номерами сайта")
args = parser.parse_args()
if args.count is not None:
limit = args.count
elif args.all or args.refresh:
limit = 10 ** 9
else:
limit = 100
conn = sqlite3.connect(DB_PATH)
ensure_schema(conn)
known_ids = {row[0] for row in conn.execute(
"SELECT site_id FROM poems WHERE site_id IS NOT NULL")}
legacy = {normalize(text): rowid for text, rowid in conn.execute(
"SELECT content, rowid FROM poems WHERE site_id IS NULL AND content IS NOT NULL")}
print(f"В базе уже {len(known_ids) + len(legacy)} стихов")
added, enriched, refreshed, empty = do_crawl(
conn, args, known_ids, legacy, limit)
total, with_author, with_date, with_likes, with_rating = conn.execute(
"SELECT COUNT(*), COUNT(author), COUNT(date), COUNT(likes), "
"COUNT(rating) FROM poems").fetchone()
conn.close()
print()
print(f"Новых: {added}, дополнено старых: {enriched}, обновлено: {refreshed}")
if empty:
print(f"Записей без текста пропущено: {empty}")
print(f"Всего в базе: {total}, с автором: {with_author}, "
f"с датой: {with_date}, с лайками: {with_likes}, "
f"с рейтингом: {with_rating}")
if __name__ == "__main__":
main()
