freedakgmail
573 lines
24 KiB
Python
573 lines
24 KiB
Python
#!/usr/bin/env python3
|
||
"""采集管道：公众号历史文章抓取 + 视频号 ASR 转写。
|
||
|
||
合规采集策略：
|
||
- 控制请求频率，每篇文章间隔 CRAWL_ARTICLE_PAUSE 秒（默认2秒）
|
||
- 列表页滚动间隔 CRAWL_LIST_PAUSE 秒（默认1.5秒）
|
||
- 最多滚动 CRAWL_MAX_SCROLLS 次（默认15次）
|
||
- 回溯天数 CRAWL_LOOKBACK_DAYS（默认7天），避免全量抓取
|
||
- 支持 Playwright 真实浏览器渲染，避免被反爬
|
||
- 视频号内容通过 ASR API 转写为文本后入库
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import re
|
||
import sqlite3
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
import urllib.error
|
||
import urllib.request
|
||
from contextlib import closing
|
||
from datetime import datetime, timedelta, timezone
|
||
from pathlib import Path
|
||
from typing import Any, Iterable
|
||
from urllib.parse import urljoin, urlsplit, urlunsplit
|
||
|
||
from dotenv import load_dotenv
|
||
|
||
BASE_DIR = Path(__file__).resolve().parent
|
||
load_dotenv(BASE_DIR / ".env")
|
||
|
||
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$")
|
||
CANYIN168_ARTICLE_RE = re.compile(r"/Article/[a-z]+/\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" and ARTICLE_PATH_RE.search(parts.path):
|
||
return urlunsplit(("https", "www.canyin88.com", parts.path, "", ""))
|
||
if host == "canyin168.com" and CANYIN168_ARTICLE_RE.search(parts.path):
|
||
return urlunsplit(("http", "www.canyin168.com", parts.path, "", ""))
|
||
return None
|
||
|
||
|
||
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"))
|
||
|
||
|
||
# ---- 频率控制器 ----
|
||
|
||
class RateLimiter:
|
||
"""简单的频率控制器，确保请求间隔不低于指定秒数。"""
|
||
|
||
def __init__(self, min_interval: float):
|
||
self.min_interval = min_interval
|
||
self._last_time = 0.0
|
||
|
||
def wait(self) -> None:
|
||
elapsed = time.time() - self._last_time
|
||
if elapsed < self.min_interval:
|
||
time.sleep(self.min_interval - elapsed)
|
||
self._last_time = time.time()
|
||
|
||
|
||
# ---- 公众号文章抓取 ----
|
||
|
||
def discover_urls(page, list_url: str, scrolls: int, pause_ms: int) -> list[str]:
|
||
"""在列表页滚动发现文章URL，控制滚动频率。"""
|
||
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, selector: str) -> str:
|
||
locator = page.locator(selector).first
|
||
return clean_text(locator.get_attribute("content")) if locator.count() else ""
|
||
|
||
|
||
def first_text(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, 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], source_type: str = "公众号", source_name: str = "") -> int:
|
||
timestamp = now_iso()
|
||
inserted = 0
|
||
for url in urls:
|
||
cursor = conn.execute(
|
||
"""
|
||
INSERT OR IGNORE INTO articles
|
||
(canonical_url, source_type, source_name, discovered_at, updated_at)
|
||
VALUES (?, ?, ?, ?, ?)
|
||
""",
|
||
(url, source_type, source_name, 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()
|
||
|
||
|
||
# ---- 视频号本地处理（参考 VIBank 方案） ----
|
||
|
||
VIDEO_SUFFIXES = {".mp4", ".mov", ".m4v", ".mkv", ".webm"}
|
||
FFMPEG = os.getenv("FFMPEG_PATH", "/usr/local/bin/ffmpeg")
|
||
FFPROBE = os.getenv("FFPROBE_PATH", "/usr/local/bin/ffprobe")
|
||
FUNASR = os.getenv("FUNASR_PATH", "/usr/local/bin/funasr")
|
||
ASR_MODEL = os.getenv("ASR_MODEL", "sensevoice")
|
||
ASR_HOTWORDS = os.getenv("ASR_HOTWORDS", "餐饮,餐饮连锁,连锁餐厅,单店模型,供应链,选址,复购,翻台率,客单价,毛利率,净利率,加盟,直营")
|
||
|
||
|
||
def load_video_config() -> dict[str, Any] | None:
|
||
"""加载视频源配置：优先从数据库 sources 表读取，降级到 JSON 文件。"""
|
||
# 优先从数据库读取
|
||
db_path = BASE_DIR / "data" / "cibank.db"
|
||
if db_path.exists():
|
||
try:
|
||
with sqlite3.connect(db_path) as conn:
|
||
rows = conn.execute(
|
||
"SELECT name, finder, download_dir FROM sources WHERE source_type='视频号' AND enabled=1"
|
||
).fetchall()
|
||
if rows:
|
||
return {
|
||
"sources": [
|
||
{"account": r[0], "finder": r[1] or "", "directory": r[2] or ""}
|
||
for r in rows
|
||
]
|
||
}
|
||
except sqlite3.Error:
|
||
pass
|
||
|
||
# 降级到 JSON 配置文件
|
||
config_path = BASE_DIR / "video_config.json"
|
||
if config_path.exists():
|
||
return json.loads(config_path.read_text(encoding="utf-8"))
|
||
legacy = BASE_DIR / "video_sources.json"
|
||
if legacy.exists():
|
||
return {"sources": [], "legacy_urls": json.loads(legacy.read_text(encoding="utf-8")).get("videos", [])}
|
||
return None
|
||
|
||
|
||
def file_fingerprint(path: Path) -> str:
|
||
stat = path.stat()
|
||
h = hashlib.sha256()
|
||
h.update(f"{stat.st_size}:{stat.st_mtime_ns}".encode())
|
||
with path.open("rb") as f:
|
||
h.update(f.read(1024 * 1024))
|
||
if stat.st_size > 2 * 1024 * 1024:
|
||
f.seek(max(0, stat.st_size - 1024 * 1024))
|
||
h.update(f.read(1024 * 1024))
|
||
return h.hexdigest()
|
||
|
||
|
||
def safe_stem(name: str) -> str:
|
||
value = re.sub(r"[_ ]?xWT\d+$", "", Path(name).stem, flags=re.I)
|
||
value = re.sub(r"[\x00-\x1f/:]", "_", value).strip(" .")
|
||
return value[:180] or "untitled"
|
||
|
||
|
||
def ffprobe_duration(path: Path) -> float:
|
||
result = subprocess.run(
|
||
[FFPROBE, "-v", "error", "-show_entries", "format=duration",
|
||
"-of", "default=noprint_wrappers=1:nokey=1", str(path)],
|
||
text=True, capture_output=True, check=True, timeout=30,
|
||
)
|
||
return float(result.stdout.strip())
|
||
|
||
|
||
def extract_audio(video: Path, audio_dir: Path) -> Path:
|
||
"""用 FFmpeg 从视频中提取 16kHz 单声道 WAV。"""
|
||
audio_dir.mkdir(parents=True, exist_ok=True)
|
||
fingerprint = file_fingerprint(video)[:10]
|
||
wav = audio_dir / f"{safe_stem(video.name)}_{fingerprint}.wav"
|
||
if wav.is_file() and wav.stat().st_size > 44:
|
||
return wav
|
||
partial = wav.with_name(wav.name + ".part")
|
||
partial.unlink(missing_ok=True)
|
||
try:
|
||
subprocess.run(
|
||
[FFMPEG, "-hide_banner", "-loglevel", "error", "-y",
|
||
"-i", str(video), "-vn", "-ac", "1", "-ar", "16000", "-c:a", "pcm_s16le", str(partial)],
|
||
check=True, timeout=1800,
|
||
)
|
||
partial.replace(wav)
|
||
return wav
|
||
finally:
|
||
partial.unlink(missing_ok=True)
|
||
|
||
|
||
def funasr_transcribe(wav_path: Path) -> str:
|
||
"""用 FunASR/SenseVoice 进行本地中文转写。"""
|
||
cmd = [FUNASR, str(wav_path), "--model", ASR_MODEL, "--language", "zh", "--output-format", "json"]
|
||
if ASR_HOTWORDS:
|
||
cmd.extend(["--hotwords", ASR_HOTWORDS])
|
||
result = subprocess.run(cmd, text=True, capture_output=True, check=True, timeout=3600)
|
||
start = result.stdout.find("{")
|
||
if start < 0:
|
||
raise RuntimeError("FunASR 未返回 JSON")
|
||
payload = json.loads(result.stdout[start:])
|
||
text = re.sub(r"<\|[^>]+\|>", "", str(payload.get("text", ""))).strip()
|
||
if not text:
|
||
raise RuntimeError("FunASR 返回空文本")
|
||
return text
|
||
|
||
|
||
def discover_local_videos(config: dict[str, Any]) -> Iterable[tuple[dict[str, str], Path]]:
|
||
"""扫描配置中的视频目录，发现本地视频文件。"""
|
||
for source in config.get("sources", []):
|
||
root = Path(source["directory"])
|
||
if not root.is_dir():
|
||
print(f"WARN 视频源目录不存在: {root}", file=sys.stderr)
|
||
continue
|
||
for path in sorted(root.iterdir()):
|
||
if path.is_file() and path.suffix.lower() in VIDEO_SUFFIXES and not path.name.startswith("."):
|
||
yield source, path
|
||
|
||
|
||
def process_local_video(source: dict[str, str], video: Path, audio_root: Path) -> dict[str, str | None | float]:
|
||
"""完整处理一个本地视频：提取音频 → ASR 转写 → 返回结构化数据。"""
|
||
duration = ffprobe_duration(video)
|
||
account = source.get("account", "未知视频号")
|
||
audio_dir = audio_root / re.sub(r"[/:]", "_", account)
|
||
wav = extract_audio(video, audio_dir)
|
||
transcript = funasr_transcribe(wav)
|
||
title = safe_stem(video.name)
|
||
content_hash = hashlib.sha256(f"{title}\n{transcript}".encode("utf-8")).hexdigest()
|
||
media_url = str(video.resolve())
|
||
media_key = hashlib.sha256(media_url.encode("utf-8")).hexdigest()[:24]
|
||
return {
|
||
"canonical_url": f"video://local/{media_key}",
|
||
"media_url": media_url,
|
||
"title": title,
|
||
"content": transcript,
|
||
"summary": transcript[:200] if transcript else "",
|
||
"author": account,
|
||
"published_at": now_iso(),
|
||
"content_hash": content_hash,
|
||
"duration": duration,
|
||
}
|
||
|
||
|
||
# ---- 主采集流程 ----
|
||
|
||
def run_crawl(args: argparse.Namespace) -> int:
|
||
db_path = Path(args.db).expanduser().resolve()
|
||
db_path.parent.mkdir(parents=True, exist_ok=True)
|
||
|
||
article_pause = float(os.getenv("CRAWL_ARTICLE_PAUSE", "2.0"))
|
||
list_pause_ms = int(float(os.getenv("CRAWL_LIST_PAUSE", "1.5")) * 1000)
|
||
max_scrolls = int(os.getenv("CRAWL_MAX_SCROLLS", "15"))
|
||
lookback_days = int(os.getenv("CRAWL_LOOKBACK_DAYS", "7"))
|
||
|
||
article_limiter = RateLimiter(article_pause)
|
||
|
||
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, source_type) VALUES (?, ?)",
|
||
(started_at, args.source_type),
|
||
).lastrowid
|
||
conn.commit()
|
||
|
||
discovered_count = inserted_count = updated_count = failed_count = 0
|
||
try:
|
||
if args.source_type == "公众号":
|
||
# 从数据库读取启用的公众号信源，降级到命令行参数
|
||
mp_sources = conn.execute(
|
||
"SELECT name, list_url FROM sources WHERE source_type='公众号' AND enabled=1 AND list_url!=''"
|
||
).fetchall()
|
||
if mp_sources:
|
||
list_urls = [(r[0], r[1]) for r in mp_sources]
|
||
else:
|
||
list_urls = [("默认", args.list_url)]
|
||
|
||
from playwright.sync_api import sync_playwright
|
||
with sync_playwright() as playwright:
|
||
browser = playwright.chromium.launch(headless=not args.headed)
|
||
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)
|
||
page = context.new_page()
|
||
try:
|
||
for source_name, list_url in list_urls:
|
||
print(f"--- 采集公众号: {source_name} ({list_url}) ---", flush=True)
|
||
urls = discover_urls(page, list_url, max_scrolls, list_pause_ms)
|
||
discovered_count += len(urls)
|
||
inserted_count += insert_discoveries(conn, urls, "公众号", source_name)
|
||
queue = pending_urls(conn, urls, args.max_retries)
|
||
cutoff = datetime.now() - timedelta(days=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
|
||
article_limiter.wait()
|
||
try:
|
||
article = extract_article(page, url)
|
||
save_success(conn, article)
|
||
updated_count += 1
|
||
print(f"[{source_name} {index}/{len(queue)}] OK {article['title']}", flush=True)
|
||
except Exception as exc:
|
||
failed_count += 1
|
||
save_failure(conn, url, exc)
|
||
print(f"[{source_name} {index}/{len(queue)}] FAIL {url}: {exc}", file=sys.stderr, flush=True)
|
||
# 更新信源的最近采集时间
|
||
conn.execute(
|
||
"UPDATE sources SET last_crawled_at=? WHERE source_type='公众号' AND name=?",
|
||
(now_iso(), source_name),
|
||
)
|
||
conn.commit()
|
||
finally:
|
||
context.close()
|
||
elif args.source_type == '视频号':
|
||
# 视频号本地处理：扫描本地视频目录 → FFmpeg 提取音频 → FunASR 转写
|
||
video_config = load_video_config()
|
||
if not video_config:
|
||
print(json.dumps({"status": "skipped", "reason": "未找到 video_config.json 或 video_sources.json"}, ensure_ascii=False))
|
||
conn.execute(
|
||
"UPDATE crawl_runs SET finished_at=?, status='skipped', message=? WHERE id=?",
|
||
(now_iso(), "未配置视频源", run_id),
|
||
)
|
||
conn.commit()
|
||
return 0
|
||
|
||
audio_root = Path(os.getenv("VIDEO_AUDIO_ROOT", str(BASE_DIR / "data" / "audio")))
|
||
video_files = list(discover_local_videos(video_config))
|
||
discovered_count = len(video_files)
|
||
|
||
for index, (source, video) in enumerate(video_files, start=1):
|
||
media_url = str(video.resolve())
|
||
existing_media = conn.execute(
|
||
"SELECT id FROM articles WHERE media_url=?", (media_url,)
|
||
).fetchone()
|
||
if existing_media:
|
||
print(f"[{index}/{discovered_count}] SKIP {video.stem}（已入库）", flush=True)
|
||
continue
|
||
article_limiter.wait()
|
||
try:
|
||
result = process_local_video(source, video, audio_root)
|
||
timestamp = now_iso()
|
||
existing = conn.execute(
|
||
"SELECT id, content_hash FROM articles WHERE canonical_url=?",
|
||
(result["canonical_url"],),
|
||
).fetchone()
|
||
if existing and existing[1] == result["content_hash"]:
|
||
print(f"[{index}/{discovered_count}] SKIP {result['title']}（已入库）", flush=True)
|
||
continue
|
||
conn.execute(
|
||
"""
|
||
INSERT INTO articles
|
||
(canonical_url, source, source_type, source_name, title, author,
|
||
published_at, summary, content, content_hash, media_url,
|
||
crawl_status, discovered_at, crawled_at, updated_at)
|
||
VALUES (?, 'video_local', '视频号', ?, ?, ?, ?, ?, ?, ?, ?, 'success', ?, ?, ?)
|
||
ON CONFLICT(canonical_url) DO UPDATE SET
|
||
title=excluded.title, summary=excluded.summary, content=excluded.content,
|
||
content_hash=excluded.content_hash, media_url=excluded.media_url,
|
||
crawled_at=excluded.crawled_at, updated_at=excluded.updated_at,
|
||
crawl_status='success', last_error=NULL
|
||
""",
|
||
(
|
||
result["canonical_url"], source.get("account", ""),
|
||
result["title"], result["author"], result["published_at"],
|
||
result["summary"], result["content"], result["content_hash"],
|
||
result["media_url"], timestamp, timestamp, timestamp,
|
||
),
|
||
)
|
||
conn.commit()
|
||
updated_count += 1
|
||
print(f"[{index}/{discovered_count}] OK {result['title']}（{result.get('duration', 0):.0f}s）", flush=True)
|
||
# 更新信源最近采集时间
|
||
conn.execute(
|
||
"UPDATE sources SET last_crawled_at=? WHERE source_type='视频号' AND name=?",
|
||
(now_iso(), source.get("account", "")),
|
||
)
|
||
conn.commit()
|
||
except Exception as exc:
|
||
failed_count += 1
|
||
print(f"[{index}/{discovered_count}] FAIL {video.name}: {exc}", file=sys.stderr, flush=True)
|
||
|
||
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, "source_type": args.source_type,
|
||
"discovered": discovered_count, "inserted": inserted_count,
|
||
"updated": updated_count, "failed": failed_count,
|
||
}, 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=str(BASE_DIR / "data" / "cibank.db"), help="SQLite 数据库路径")
|
||
parser.add_argument("--list-url", default=DEFAULT_LIST_URL, help="公众号列表页URL")
|
||
parser.add_argument("--source-type", choices=["公众号", "视频号"], default="公众号", 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_crawl(build_parser().parse_args()))
|