\s*
(.*?)',
html_text,
re.I | re.S
)
items = []
for date_text, href, title in matches[:MAX_ITEMS]:
items.append({
"date_text": clean_text(re.sub(r"<[^>]+>", " ", date_text)),
"source_url": urljoin(LIST_URL, href),
"title": clean_text(re.sub(r"<[^>]+>", " ", title)),
})
return items
def parse_detail(session: requests.Session, url: str) -> dict:
resp = session.get(url, timeout=30)
resp.raise_for_status()
html_text = resp.text
heading_matches = re.findall(r"
]*>(.*?)", html_text, re.I | re.S)
title = ""
for raw in heading_matches:
text = clean_text(re.sub(r"<[^>]+>", " ", raw))
if text and text != "其他新聞":
title = text
break
paragraphs = []
for raw in re.findall(r"]*>([\s\S]*?)", html_text, re.I):
text = clean_text(re.sub(r"<[^>]+>", " ", raw))
if len(text) > 8:
paragraphs.append(text)
images = []
for src in re.findall(r']+src="([^"]+res\.hkjc\.com[^"]+)"', html_text, re.I):
full = urljoin(url, src)
if full not in images:
images.append(full)
published_text = ""
if paragraphs and re.match(r"^\d{2}/\d{2}/\d{4}", paragraphs[0]):
published_text = paragraphs.pop(0)
content_parts = [f"{text}
" for text in paragraphs]
for image in images[:3]:
content_parts.append(f'
')
desc = paragraphs[0][:255] if paragraphs else title[:255]
return {
"title": title[:255],
"desc": desc,
"content": "".join(content_parts),
"image": images[0] if images else "",
"published_text": published_text[:64],
"source_url": url,
}
def sync_to_db(conn, articles: list[dict]) -> tuple[int, int, int]:
if not articles:
return 0, 0, 0
urls = [a["source_url"] for a in articles if a.get("source_url")]
with conn.cursor() as cur:
placeholders = ",".join(["%s"] * len(urls))
cur.execute(
f"SELECT id, source_url, content, is_show FROM la_article WHERE source_url IN ({placeholders}) AND cid = %s",
urls + [ARTICLE_CID]
)
existing = {row["source_url"]: row for row in cur.fetchall()}
titles = [a["title"][:255] for a in articles if a.get("title")]
existing_titles = set()
if titles:
with conn.cursor() as cur:
t_placeholders = ",".join(["%s"] * len(titles))
cur.execute(
f"SELECT title FROM la_article WHERE title IN ({t_placeholders}) AND cid = %s",
titles + [ARTICLE_CID]
)
existing_titles = {row["title"] for row in cur.fetchall()}
now = int(time.time())
new_count = 0
update_count = 0
for article in articles:
title = article.get("title", "")
source_url = article.get("source_url", "")
if not title or not source_url:
continue
ext_json = json.dumps({
"source": "hkjc_racingnews",
"list_url": LIST_URL,
}, ensure_ascii=False)
if source_url in existing:
existing_row = existing[source_url]
next_is_show = existing_row["is_show"] if has_publishable_article_content(existing_row.get("content", "")) else 0
with conn.cursor() as cur:
cur.execute(
"""UPDATE la_article SET
title=%s, `desc`=%s, abstract=%s, image=%s, author=%s,
published_at=%s, category=%s, content=%s, ext=%s, is_show=%s, update_time=%s
WHERE id=%s""",
(
title[:255],
(article.get("desc") or "")[:255],
(article.get("desc") or "")[:500],
(article.get("image") or "")[:128],
SOURCE_NAME,
(article.get("published_at") or "")[:64],
"lottery,hk",
article.get("content") or "",
ext_json,
next_is_show,
now,
existing_row["id"],
)
)
update_count += 1
else:
if title[:255] in existing_titles:
continue
with conn.cursor() as cur:
cur.execute(
"""INSERT INTO la_article
(cid, title, `desc`, abstract, image, author, published_at,
category, source_url, content, ext, is_show, sort,
click_virtual, click_actual, create_time, update_time)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)""",
(
ARTICLE_CID,
title[:255],
(article.get("desc") or "")[:255],
(article.get("desc") or "")[:500],
(article.get("image") or "")[:128],
SOURCE_NAME,
(article.get("published_at") or "")[:64],
"lottery,hk",
source_url[:500],
article.get("content") or "",
ext_json,
resolve_crawled_article_is_show(article.get("content") or ""),
0,
0,
0,
article.get("published_ts") or now,
now,
)
)
existing[source_url] = {"id": cur.lastrowid, "content": article.get("content") or "", "is_show": resolve_crawled_article_is_show(article.get("content") or "")}
new_count += 1
existing_titles.add(title[:255])
return len(articles), new_count, update_count
def run(db_config: dict):
session = requests.Session()
session.headers.update({
"User-Agent": "Mozilla/5.0 (compatible; SportEra/1.0)",
"Accept": "text/html,*/*",
})
list_items = fetch_list(session)
print(f"HKJC 资讯列表返回 {len(list_items)} 条")
articles = []
for item in list_items:
detail = parse_detail(session, item["source_url"])
published_ts, published_at = parse_hk_date(detail.get("published_text") or item.get("date_text", ""))
articles.append({
"title": detail.get("title") or item["title"],
"desc": detail.get("desc", ""),
"content": detail.get("content", ""),
"image": detail.get("image", ""),
"published_ts": published_ts,
"published_at": published_at,
"source_url": item["source_url"],
})
if not articles:
print("无可写入数据，跳过")
return {"success": True, "candidate_count": 0, "saved_count": 0, "summary_text": "无可写入数据，跳过"}
conn = pymysql.connect(**db_config, cursorclass=pymysql.cursors.DictCursor)
try:
total, new_count, update_count = sync_to_db(conn, articles)
conn.commit()
summary = f"共 {total} 条, 新增 {new_count}, 更新 {update_count}"
print(summary)
return {"success": True, "candidate_count": total, "saved_count": new_count + update_count, "summary_text": summary}
finally:
conn.close()