初始化红餐观察库项目
This commit is contained in:
freedakgmail
@@ -0,0 +1,301 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Incrementally collect Hongcan articles into SQLite using a real browser."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import hashlib
|
||||
import json
|
||||
import re
|
||||
import sqlite3
|
||||
import sys
|
||||
import time
|
||||
from contextlib import closing
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from pathlib import Path
|
||||
from urllib.parse import urljoin, urlsplit, urlunsplit
|
||||
|
||||
from playwright.sync_api import BrowserContext, Page, sync_playwright
|
||||
|
||||
|
||||
DEFAULT_LIST_URL = "https://m.canyin88.com/zixun/"
|
||||
ARTICLE_PATH_RE = re.compile(r"/zixun/\d{4}/\d{1,2}/\d{1,2}/\d+\.html$")
|
||||
DATE_RE = re.compile(r"(20\d{2})[-/.年](\d{1,2})[-/.月](\d{1,2})(?:日)?(?:\s+(\d{1,2}):(\d{2})(?::(\d{2}))?)?")
|
||||
|
||||
|
||||
def now_iso() -> str:
|
||||
return datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds")
|
||||
|
||||
|
||||
def normalize_url(raw_url: str, base_url: str = DEFAULT_LIST_URL) -> str | None:
|
||||
absolute = urljoin(base_url, raw_url.strip())
|
||||
parts = urlsplit(absolute)
|
||||
host = parts.netloc.lower().removeprefix("m.").removeprefix("www.")
|
||||
if host != "canyin88.com" or not ARTICLE_PATH_RE.search(parts.path):
|
||||
return None
|
||||
return urlunsplit(("https", "www.canyin88.com", parts.path, "", ""))
|
||||
|
||||
|
||||
def clean_text(value: str | None) -> str:
|
||||
return re.sub(r"\s+", " ", value or "").strip().lstrip("\ufeff")
|
||||
|
||||
|
||||
def parse_date(value: str | None) -> str | None:
|
||||
if not value:
|
||||
return None
|
||||
match = DATE_RE.search(value)
|
||||
if not match:
|
||||
return None
|
||||
year, month, day, hour, minute, second = match.groups()
|
||||
dt = datetime(
|
||||
int(year), int(month), int(day), int(hour or 0), int(minute or 0), int(second or 0)
|
||||
)
|
||||
return dt.isoformat(timespec="seconds")
|
||||
|
||||
|
||||
def date_from_article_url(url: str) -> datetime | None:
|
||||
match = re.search(r"/zixun/(20\d{2})/(\d{1,2})/(\d{1,2})/", url)
|
||||
if not match:
|
||||
return None
|
||||
return datetime(*(int(part) for part in match.groups()))
|
||||
|
||||
|
||||
def init_db(conn: sqlite3.Connection, schema_path: Path) -> None:
|
||||
conn.executescript(schema_path.read_text(encoding="utf-8"))
|
||||
|
||||
|
||||
def discover_urls(page: Page, list_url: str, scrolls: int, pause_ms: int) -> list[str]:
|
||||
page.goto(list_url, wait_until="domcontentloaded", timeout=60_000)
|
||||
page.wait_for_timeout(2_000)
|
||||
found: set[str] = set()
|
||||
stagnant = 0
|
||||
for _ in range(scrolls + 1):
|
||||
hrefs = page.locator("a[href]").evaluate_all("els => els.map(e => e.href)")
|
||||
before = len(found)
|
||||
for href in hrefs:
|
||||
normalized = normalize_url(href, list_url)
|
||||
if normalized:
|
||||
found.add(normalized)
|
||||
stagnant = stagnant + 1 if len(found) == before else 0
|
||||
if stagnant >= 3:
|
||||
break
|
||||
page.evaluate("window.scrollTo(0, document.body.scrollHeight)")
|
||||
page.wait_for_timeout(pause_ms)
|
||||
return sorted(found, reverse=True)
|
||||
|
||||
|
||||
def meta(page: Page, selector: str) -> str:
|
||||
locator = page.locator(selector).first
|
||||
return clean_text(locator.get_attribute("content")) if locator.count() else ""
|
||||
|
||||
|
||||
def first_text(page: Page, selectors: list[str]) -> str:
|
||||
for selector in selectors:
|
||||
locator = page.locator(selector).first
|
||||
if locator.count():
|
||||
text = clean_text(locator.inner_text(timeout=3_000))
|
||||
if text:
|
||||
return text
|
||||
return ""
|
||||
|
||||
|
||||
def extract_article(page: Page, url: str) -> dict[str, str | None]:
|
||||
page.goto(url, wait_until="domcontentloaded", timeout=60_000)
|
||||
page.wait_for_timeout(800)
|
||||
document_title = clean_text(page.title())
|
||||
document_title = re.sub(r"(?:[_-]红餐网.*)$", "", document_title).strip()
|
||||
title = meta(page, 'meta[property="og:title"]') or document_title
|
||||
if not title or title == "相关推荐":
|
||||
title = first_text(page, [".title", "h1"])
|
||||
summary = (
|
||||
meta(page, 'meta[name="description"]')
|
||||
or meta(page, 'meta[property="og:description"]')
|
||||
)
|
||||
author = (
|
||||
meta(page, 'meta[name="author"]')
|
||||
or first_text(page, [".author", ".source", "[class*=author]"])
|
||||
)
|
||||
published_raw = (
|
||||
meta(page, 'meta[property="article:published_time"]')
|
||||
or first_text(page, ["time", ".time", "[class*=time]", "[class*=date]"])
|
||||
or page.locator("body").inner_text(timeout=5_000)[:1000]
|
||||
)
|
||||
content = first_text(
|
||||
page,
|
||||
[
|
||||
"article",
|
||||
".article-content",
|
||||
".article_content",
|
||||
".content",
|
||||
"[class*=detail-content]",
|
||||
"[class*=article] [class*=content]",
|
||||
],
|
||||
)
|
||||
if not content:
|
||||
paragraphs = page.locator("p").all_inner_texts()
|
||||
content = "\n\n".join(clean_text(p) for p in paragraphs if len(clean_text(p)) >= 20)
|
||||
if not title or len(content) < 80:
|
||||
raise ValueError(f"article extraction incomplete: title={bool(title)}, content_length={len(content)}")
|
||||
published_at = parse_date(published_raw)
|
||||
url_date = date_from_article_url(url)
|
||||
if url_date and (
|
||||
not published_at or datetime.fromisoformat(published_at).date() != url_date.date()
|
||||
):
|
||||
published_at = url_date.isoformat(timespec="seconds")
|
||||
content_hash = hashlib.sha256(f"{title}\n{content}".encode("utf-8")).hexdigest()
|
||||
return {
|
||||
"canonical_url": url,
|
||||
"title": title,
|
||||
"author": author or None,
|
||||
"published_at": published_at,
|
||||
"summary": summary or None,
|
||||
"content": content,
|
||||
"content_hash": content_hash,
|
||||
}
|
||||
|
||||
|
||||
def pending_urls(conn: sqlite3.Connection, discovered: list[str], max_retries: int) -> list[str]:
|
||||
rows = conn.execute(
|
||||
"SELECT canonical_url FROM articles WHERE crawl_status != 'success' AND retry_count < ?",
|
||||
(max_retries,),
|
||||
).fetchall()
|
||||
return list(dict.fromkeys(discovered + [row[0] for row in rows]))
|
||||
|
||||
|
||||
def insert_discoveries(conn: sqlite3.Connection, urls: list[str]) -> int:
|
||||
timestamp = now_iso()
|
||||
inserted = 0
|
||||
for url in urls:
|
||||
cursor = conn.execute(
|
||||
"""
|
||||
INSERT OR IGNORE INTO articles
|
||||
(canonical_url, discovered_at, updated_at)
|
||||
VALUES (?, ?, ?)
|
||||
""",
|
||||
(url, timestamp, timestamp),
|
||||
)
|
||||
inserted += cursor.rowcount
|
||||
conn.commit()
|
||||
return inserted
|
||||
|
||||
|
||||
def save_success(conn: sqlite3.Connection, article: dict[str, str | None]) -> None:
|
||||
timestamp = now_iso()
|
||||
conn.execute(
|
||||
"""
|
||||
UPDATE articles SET
|
||||
title = ?, author = ?, published_at = ?, summary = ?, content = ?,
|
||||
content_hash = ?, crawl_status = 'success', last_error = NULL,
|
||||
crawled_at = ?, updated_at = ?
|
||||
WHERE canonical_url = ?
|
||||
""",
|
||||
(
|
||||
article["title"], article["author"], article["published_at"], article["summary"],
|
||||
article["content"], article["content_hash"], timestamp, timestamp,
|
||||
article["canonical_url"],
|
||||
),
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
|
||||
def save_failure(conn: sqlite3.Connection, url: str, error: Exception) -> None:
|
||||
conn.execute(
|
||||
"""
|
||||
UPDATE articles SET
|
||||
crawl_status = 'failed', retry_count = retry_count + 1,
|
||||
last_error = ?, updated_at = ?
|
||||
WHERE canonical_url = ?
|
||||
""",
|
||||
(str(error)[:1000], now_iso(), url),
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
|
||||
def new_context(playwright, headless: bool) -> tuple[BrowserContext, Page]:
|
||||
browser = playwright.chromium.launch(headless=headless)
|
||||
context = browser.new_context(
|
||||
locale="zh-CN",
|
||||
timezone_id="Asia/Shanghai",
|
||||
viewport={"width": 1280, "height": 900},
|
||||
user_agent=(
|
||||
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
|
||||
"AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0 Safari/537.36"
|
||||
),
|
||||
)
|
||||
context.set_default_timeout(15_000)
|
||||
return context, context.new_page()
|
||||
|
||||
|
||||
def run(args: argparse.Namespace) -> int:
|
||||
base_dir = Path(__file__).resolve().parent
|
||||
db_path = Path(args.db).expanduser().resolve()
|
||||
db_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
with closing(sqlite3.connect(db_path)) as conn:
|
||||
init_db(conn, base_dir / "schema.sql")
|
||||
started_at = now_iso()
|
||||
run_id = conn.execute(
|
||||
"INSERT INTO crawl_runs(started_at) VALUES (?)", (started_at,)
|
||||
).lastrowid
|
||||
conn.commit()
|
||||
discovered_count = inserted_count = updated_count = failed_count = 0
|
||||
try:
|
||||
with sync_playwright() as playwright:
|
||||
context, page = new_context(playwright, args.headed is False)
|
||||
try:
|
||||
urls = discover_urls(page, args.list_url, args.scrolls, args.pause_ms)
|
||||
discovered_count = len(urls)
|
||||
inserted_count = insert_discoveries(conn, urls)
|
||||
queue = pending_urls(conn, urls, args.max_retries)
|
||||
cutoff = datetime.now() - timedelta(days=args.lookback_days)
|
||||
for index, url in enumerate(queue, start=1):
|
||||
path_date = parse_date(url.replace("/", "-"))
|
||||
if path_date and datetime.fromisoformat(path_date) < cutoff:
|
||||
continue
|
||||
try:
|
||||
article = extract_article(page, url)
|
||||
save_success(conn, article)
|
||||
updated_count += 1
|
||||
print(f"[{index}/{len(queue)}] OK {article['title']}")
|
||||
except Exception as exc:
|
||||
failed_count += 1
|
||||
save_failure(conn, url, exc)
|
||||
print(f"[{index}/{len(queue)}] FAIL {url}: {exc}", file=sys.stderr)
|
||||
time.sleep(args.article_pause)
|
||||
finally:
|
||||
context.close()
|
||||
status = "success" if failed_count == 0 else "partial"
|
||||
message = json.dumps({"db": str(db_path)}, ensure_ascii=False)
|
||||
except Exception as exc:
|
||||
status, message = "failed", str(exc)[:1000]
|
||||
print(f"crawl failed: {exc}", file=sys.stderr)
|
||||
conn.execute(
|
||||
"""
|
||||
UPDATE crawl_runs SET finished_at=?, status=?, discovered_count=?,
|
||||
inserted_count=?, updated_count=?, failed_count=?, message=?
|
||||
WHERE id=?
|
||||
""",
|
||||
(now_iso(), status, discovered_count, inserted_count, updated_count, failed_count, message, run_id),
|
||||
)
|
||||
conn.commit()
|
||||
print(json.dumps({
|
||||
"status": status, "discovered": discovered_count, "inserted": inserted_count,
|
||||
"updated": updated_count, "failed": failed_count, "database": str(db_path),
|
||||
}, ensure_ascii=False))
|
||||
return 0 if status in {"success", "partial"} else 1
|
||||
|
||||
|
||||
def build_parser() -> argparse.ArgumentParser:
|
||||
parser = argparse.ArgumentParser(description="红餐网文章每日增量采集入库")
|
||||
parser.add_argument("--db", default="hongcan.db", help="SQLite 数据库路径")
|
||||
parser.add_argument("--list-url", default=DEFAULT_LIST_URL)
|
||||
parser.add_argument("--lookback-days", type=int, default=7, help="回溯天数")
|
||||
parser.add_argument("--scrolls", type=int, default=12, help="列表页最大下拉次数")
|
||||
parser.add_argument("--pause-ms", type=int, default=1200, help="列表页下拉等待毫秒")
|
||||
parser.add_argument("--article-pause", type=float, default=0.8, help="文章间隔秒数")
|
||||
parser.add_argument("--max-retries", type=int, default=3)
|
||||
parser.add_argument("--headed", action="store_true", help="显示浏览器，便于排查验证码")
|
||||
return parser
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(run(build_parser().parse_args()))
|
||||
Reference in New Issue
Block a user