"""Supplement sparse noodle topics with archived and public original articles. Search results are discovery only. Short community posts keep their full actual body; no minimum word count is padded with recommendations or page furniture. """ from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime, timezone import hashlib import importlib.util import json from pathlib import Path import re import time import urllib.parse from lxml import html ROOT = Path(__file__).resolve().parents[1] ORIGINAL_WORK = ROOT / 'work/five-topic-collection-20261110' WORK = ORIGINAL_WORK / 'supplement-01' PUBLIC = WORK / 'public-sources' SEARCH = WORK / 'search' for directory in [WORK, PUBLIC, SEARCH]: directory.mkdir(parents=True, exist_ok=True) spec = importlib.util.spec_from_file_location('base_public', ROOT / 'scripts/collect-five-topic-public.py') base = importlib.util.module_from_spec(spec) spec.loader.exec_module(base) base.PUBLIC = PUBLIC POST = re.compile(r'https://post\.smzdm\.com/(?:talk/)?p/([a-z0-9]+)/?$') QUERIES = [ ('wellness', '菌菇 方便面 测评 site:post.smzdm.com'), ('premium', '拉面说 牛肉 评测 site:post.smzdm.com'), ('stockpile', '方便面 囤货 site:post.smzdm.com'), ('wellness', '蘑菇 鲜蔬 方便面 site:post.smzdm.com'), ('premium', '满汉大餐 试吃 site:post.smzdm.com'), ('stockpile', '泡面 家里 常备 site:post.smzdm.com'), ('wellness', '五谷道场 非油炸 测评 site:post.smzdm.com'), ('premium', '速达面馆 牛肉 试吃 site:post.smzdm.com'), ('stockpile', '泡面 整箱 囤 site:post.smzdm.com'), ('novelty', '方便面 冬阴功 尝鲜 site:post.smzdm.com'), ('wellness', '方便面 枸杞 山药 site:post.smzdm.com'), ('premium', '拉面说 满汉宴 真肉 site:post.smzdm.com'), ('stockpile', '方便面 囤货 保质期 site:post.smzdm.com'), ('novelty', '方便面 藤椒 香菜 测评 site:post.smzdm.com'), ('wellness', '方便面 低钠 减盐 site:post.smzdm.com'), ('stockpile', '泡面 家庭 囤货 组合 site:post.smzdm.com'), ] STATE = {'schema': 'ksf.five-topic.supplement-collection.v1', 'startedAt': base.now(), 'status': 'running', 'searches': [], 'sources': [], 'rejected': [], 'productionWrites': 0} def persist(): STATE['updatedAt'] = base.now() base.save(WORK / 'collection-result.json', STATE) def canonical(url): parsed = urllib.parse.urlparse(url) if parsed.hostname != 'post.smzdm.com': return None value = 'https://post.smzdm.com' + parsed.path return value if POST.fullmatch(value) else None def decode(raw, receipt): declared = re.search(r'charset\s*=\s*([^;\s]+)', receipt.get('contentType', ''), re.I) encoding = declared.group(1).strip('"\'') if declared else 'utf8' try: return raw.decode(encoding), encoding except (LookupError, UnicodeDecodeError): try: return raw.decode('utf8'), 'utf8' except UnicodeDecodeError: return raw.decode('gb18030'), 'gb18030' def extract(raw, receipt): text, encoding = decode(raw, receipt) root = html.fromstring(text) title = re.sub(r'\s+', ' ', root.xpath('string(//title)')).strip() if re.search(r'验证|验证码|页面不存在|访问异常|系统繁忙|404', title): raise ValueError('VERIFICATION_OR_ERROR_PAGE') articles = root.xpath('//article') if len(articles) != 1: raise ValueError('ORIGINAL_ARTICLE_NOT_UNIQUE') article = articles[0] heading = article.xpath('.//h1') if heading: title = re.sub(r'\s+', ' ', heading[0].text_content()).strip() or title selector = '//article' body = base.paragraphs(article) # SMZDM talk posts can be short but complete. Exclude a title-only shell. body_only = body.removeprefix(title).strip() # This only rejects an empty shell; semantic review decides usefulness. if len(body_only) < 50: raise ValueError('INSUFFICIENT_ACTUAL_ARTICLE_BODY') if not re.search(r'方便面|泡面|即食面|速食面|拌面|火鸡面|拉面说|速达面馆|满汉大餐|满汉宴', title + '\n' + body): raise ValueError('NOT_PACKAGED_NOODLE_EXPERIENCE') dates = root.xpath('//meta[@property="article:published_time" or @name="pubdate" or @name="weibo:article:create_at"]/@content') if not dates: match = re.search(r'\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}(?::\d{2})?', article.text_content()[:1200]) dates = [match.group(0)] if match else [] published = dates[0] if dates else None if published: match = re.match(r'(\d{4}-\d{2}-\d{2})', published) if match and match.group(1) > datetime.now().strftime('%Y-%m-%d'): raise ValueError('FUTURE_PUBLICATION_DATE') return title, body, selector, published, encoding def make_source(raw, receipt, candidate, reused): url = canonical(receipt['finalUrl']) if not url: raise ValueError('NOT_CANONICAL_SMZDM_ORIGINAL') title, body, selector, published, encoding = extract(raw, receipt) matches = [] for topic in base.PLAN['topics']: hits = [term for term in topic['keywords'] if term in title + '\n' + body] if hits: matches.append({'topic': topic['id'], 'matchedTerms': hits}) return {'schema': 'ksf.five-topic.public-source.v1', 'platform': 'smzdm', 'sourceType': 'public_article_not_ecommerce_review', 'sourceId': 'url-sha256:' + base.digest(url.encode()), 'url': url, 'title': title, 'body': body, 'bodySha256': base.digest(body.encode()), 'publishedAtRaw': published, 'collectedAt': receipt['collectedAt'], 'httpReceipt': receipt, 'sourceEncoding': encoding, 'bodyExtractionSelector': selector, 'topics': matches, 'searchDiscovery': candidate, 'previouslyArchivedOriginalReused': reused, 'publishedToDatabase': False, 'limits': ['公开社区作者原文，不等同电商购买评价或独立消费者', '作者功效、营养、原料、工艺及价格声明未在本轮独立验证', '保留原日期，不把历史原文当作当前趋势']} def archived_sources(existing): rows = {} for file in sorted((ORIGINAL_WORK / 'public-sources').glob('*.http.json')): receipt = json.loads(file.read_bytes()) url = canonical(receipt['finalUrl']) if not url or url in existing or receipt['httpStatus'] != 200: continue raw = (file.parent / receipt['responseFile']).read_bytes() if base.digest(raw) != receipt['sha256']: raise ValueError('ARCHIVED_ORIGINAL_SHA_MISMATCH') key = 'archive-' + base.digest(url.encode())[:24] (PUBLIC / (key + '.body')).write_bytes(raw) copied = {**receipt, 'responseFile': key + '.body', 'previousReceipt': str(file.relative_to(ROOT))} base.save(PUBLIC / (key + '.http.json'), copied) try: record = make_source(raw, copied, {'type': 'prior_preserved_original'}, True) base.save(PUBLIC / (key + '.source.json'), record) rows[url] = record except ValueError as error: STATE['rejected'].append({'url': url, 'reason': str(error), 'archiveReused': True}) return rows def search(index, topic, query): key = 'query-' + str(index + 1).zfill(2) url = 'https://www.sogou.com/web?' + urllib.parse.urlencode({'query': query}) raw, receipt = base.fetch(url, key, SEARCH) decoded, _ = decode(raw, receipt) root = html.fromstring(decoded) title = root.xpath('string(//title)') if receipt['httpStatus'] != 200 or re.search(r'验证|验证码|访问异常', title): raise ValueError('SEARCH_UNAVAILABLE_OR_VERIFICATION') rows = [] for heading in root.xpath('//h3'): links = heading.xpath('.//a/@href') if not links: continue text = re.sub(r'\s+', ' ', heading.text_content()).strip() if re.search(r'高清大图|做法步骤|菜谱大全|价格表|批发|青团', text): continue rows.append({'topic': topic, 'query': query, 'title': text, 'url': urllib.parse.urljoin(receipt['finalUrl'], links[0]), 'searchReceipt': key + '.http.json'}) base.save(SEARCH / (key + '.results.json'), {'query': query, 'receipt': receipt, 'results': rows}) STATE['searches'].append({'query': query, 'topic': topic, 'candidateCount': len(rows), 'sha256': receipt['sha256']}) return rows def collect(candidate): key = base.digest(candidate['url'].encode())[:24] cached = PUBLIC / (key + '.source.json') if cached.exists(): result = json.loads(cached.read_bytes()) raw = (PUBLIC / result['httpReceipt']['responseFile']).read_bytes() assert base.digest(raw) == result['httpReceipt']['sha256'] return result raw, receipt = base.fetch(candidate['url'], key, PUBLIC) if receipt['httpStatus'] != 200: raise ValueError('HTTP_' + str(receipt['httpStatus'])) raw, receipt = base.resolve(raw, receipt, key) if receipt['httpStatus'] != 200: raise ValueError('ORIGINAL_HTTP_' + str(receipt['httpStatus'])) result = make_source(raw, receipt, candidate, False) base.save(cached, result) return result def main(): prior = json.loads((ORIGINAL_WORK / 'reviewed-public-sources.json').read_bytes()) existing = {r['url'] for r in prior['selected']} unique = archived_sources(existing) STATE['sources'] = list(unique.values()) persist() print(json.dumps({'stage': 'archive', 'recoveredOriginalBodies': len(unique)}, ensure_ascii=False), flush=True) candidates = {} for index, (topic, query) in enumerate(QUERIES): try: for row in search(index, topic, query): candidates.setdefault(row['url'], row) except Exception as error: STATE['searches'].append({'query': query, 'topic': topic, 'error': str(error)}) if str(error) == 'SEARCH_UNAVAILABLE_OR_VERIFICATION': STATE['searchStopReason'] = str(error) break persist() print(json.dumps({'stage': 'search', 'done': index + 1, 'candidates': len(candidates)}, ensure_ascii=False), flush=True) time.sleep(1.3) STATE['candidateCount'] = len(candidates) with ThreadPoolExecutor(max_workers=2) as pool: tasks = {pool.submit(collect, row): row for row in candidates.values()} for task in as_completed(tasks): row = tasks[task] try: result = task.result() if result['url'] not in existing: unique.setdefault(result['url'], result) except Exception as error: STATE['rejected'].append({'candidate': row, 'reason': str(error)}) STATE['sources'] = list(unique.values()) persist() print(json.dumps({'stage': 'body', 'originalBodies': len(unique), 'rejected': len(STATE['rejected'])}, ensure_ascii=False), flush=True) STATE['status'] = 'collection-complete-awaiting-semantic-review' STATE['finishedAt'] = base.now() STATE['uniqueSources'] = len(unique) persist() review = [{'index': index, 'url': row['url'], 'title': row['title'], 'date': row['publishedAtRaw'], 'body': row['body'], 'keywordTopics': row['topics']} for index, row in enumerate(STATE['sources'])] base.save(WORK / 'review-candidates.json', review) print(json.dumps({'status': STATE['status'], 'uniqueSources': len(unique)}, ensure_ascii=False), flush=True) if __name__ == '__main__': main()