"""Supplement sparse topics using precise original pages and publisher search. Reuse preserved discovery receipts, stop a host when a challenge page appears, and never count search snippets or recommended content as original evidence. """ from datetime import datetime 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] PARENT = ROOT / 'work/five-topic-collection-20261110' WORK = PARENT / 'supplement-02' PUBLIC = WORK / 'sources' SEARCH = WORK / 'search' for directory in [WORK, PUBLIC, SEARCH]: directory.mkdir(parents=True, exist_ok=True) spec = importlib.util.spec_from_file_location('public_base_round3', ROOT / 'scripts/collect-five-topic-public.py') base = importlib.util.module_from_spec(spec) spec.loader.exec_module(base) spec = importlib.util.spec_from_file_location('industry_base_round3', ROOT / 'scripts/collect-five-topic-industry-supplement.py') industry = importlib.util.module_from_spec(spec) spec.loader.exec_module(industry) QUERIES = ['中式养生', '药食同源', '草本', '拉面说', '保质期', '五连包', '组合装', '家庭装'] DISCOVERY = [('wellness', '人参 老母鸡 方便面 测评'), ('stockpile', '方便面 囤货 保质期 测评'), ('wellness', '当归 花雕鸡 面 统一'), ('premium', '速达面馆 牛肉 测评')] STATE = {'schema': 'ksf.five-topic.round3-collection.v1', 'startedAt': base.now(), 'status': 'running', 'searches': [], 'sources': [], 'rejected': [], 'hostStops': {}, 'productionWrites': 0} def persist(): STATE['updatedAt'] = base.now() base.save(WORK / 'collection-result.json', STATE) def existing_urls(): rows = [] for name in ['five-topics-public-01', 'five-topics-public-02']: rows.extend(json.loads((ROOT / 'work/database-enrichment-20261110/batches' / name / 'expected-records.json').read_bytes())) return {row['url'] for row in rows} def capture(url, name, target): # Reuse a prior exact response if present in this round's initial probes. for receipt_file in (WORK / 'probes').glob('*.http.json'): receipt = json.loads(receipt_file.read_bytes()) if receipt['requestedUrl'] == url and receipt['httpStatus'] == 200: raw = (receipt_file.parent / receipt['responseFile']).read_bytes() assert base.digest(raw) == receipt['sha256'] copied = {**receipt, 'responseFile': name + '.body', 'reusedReceipt': str(receipt_file.relative_to(ROOT))} (target / copied['responseFile']).write_bytes(raw) base.save(target / (name + '.http.json'), copied) return raw, copied return base.fetch(url, name, target) def challenge(raw, receipt, root): title = root.xpath('string(//title)') if re.search(r'/geetest|captcha|/verify', receipt.get('finalUrl', ''), re.I): return True if receipt['httpStatus'] in (202, 403, 412, 429) or re.search(r'安全验证|访问验证|验证码|访问异常|验证中心', title): return True if len(raw) < 10000 and re.search(rb'aliyun|captcha|acw_sc__|var seqid\s*=', raw, re.I): return True return False def collect(url, discovery): host = urllib.parse.urlparse(url).hostname if host in STATE['hostStops']: raise ValueError('HOST_STOPPED_AFTER_CHALLENGE') key = base.digest(url.encode())[:24] raw, receipt = capture(url, key, PUBLIC) root = html.fromstring(raw.decode('utf8', errors='replace')) if challenge(raw, receipt, root): STATE['hostStops'][host] = {'reason': 'VERIFICATION_RESPONSE', 'url': url, 'httpStatus': receipt['httpStatus'], 'sha256': receipt['sha256']} raise ValueError('VERIFICATION_RESPONSE') if receipt['httpStatus'] != 200: raise ValueError('HTTP_' + str(receipt['httpStatus'])) date = None if host == 'post.smzdm.com': match = re.fullmatch(r'https://post\.smzdm\.com/(?:talk/)?p/([a-z0-9]+)/', receipt['finalUrl']) if not match: raise ValueError('CANONICAL_ORIGINAL_POST_REQUIRED') platform, pid = 'smzdm', match.group(1) selector = '//article' nodes = root.xpath(selector) dates = root.xpath('//meta[@property="article:published_time"]/@content') if not dates and len(nodes) == 1: match_date = re.search(r'\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}(?::\d{2})?', nodes[0].text_content()[:1000]) dates = [match_date.group(0)] if match_date else [] date = dates[0] if dates else None elif host == 'sspai.com': match = re.fullmatch(r'https://sspai\.com/post/(\d+)', receipt['finalUrl']) if not match: raise ValueError('CANONICAL_ORIGINAL_POST_REQUIRED') platform, pid = 'sspai', match.group(1) selector = '//*[contains(concat(" ",normalize-space(@class)," ")," article-body ")]' nodes = root.xpath(selector) dates = root.xpath('//*[contains(concat(" ",normalize-space(@class)," ")," article__header__date ")]/text()') date = dates[0].strip() if dates else None elif host == 'www.foodaily.com': title, body, selector, date, encoding = industry.extract(raw, receipt) platform, pid = 'foodaily', urllib.parse.urlparse(url).path.split('/')[-1] nodes = root.xpath(selector) else: raise ValueError('PUBLISHER_NOT_IN_CONTROLLED_EXTRACTORS') if len(nodes) != 1: raise ValueError('UNIQUE_ORIGINAL_BODY_REQUIRED') body = base.paragraphs(nodes[0]) if len(body) < 60: raise ValueError('INSUFFICIENT_SOURCE_BODY') if not re.search(r'方便面|泡面|速食面|拉面说|速达面馆|满汉|即食面', body): raise ValueError('NOT_PACKAGED_NOODLE_CONTENT') title = root.xpath('string(//title)').strip() h1 = root.xpath('//article//h1') if h1: title = re.sub(r'\s+', ' ', h1[0].text_content()).strip() or title if not date: raise ValueError('ORIGINAL_DATE_NOT_LOCATED') date_match = re.search(r'(\d{4})[-/.年](\d{1,2})[-/.月](\d{1,2})', date) if date_match: normal = '-'.join([date_match.group(1), date_match.group(2).zfill(2), date_match.group(3).zfill(2)]) if normal > datetime.now().strftime('%Y-%m-%d'): raise ValueError('FUTURE_PUBLICATION_DATE') row = {'schema': 'ksf.five-topic.public-source.v1', 'platform': platform, 'postId': pid, 'sourceId': platform + ':' + pid, 'sourceType': 'public_industry_article_not_consumer_review' if platform == 'foodaily' else 'public_article_not_ecommerce_review', 'url': receipt['finalUrl'], 'title': title, 'body': body, 'bodySha256': base.digest(body.encode()), 'publishedAtRaw': date, 'collectedAt': receipt['collectedAt'], 'httpReceipt': receipt, 'sourceEncoding': 'utf8', 'bodyExtractionSelector': selector, 'collectionDirectory': 'sources', 'searchDiscovery': discovery, 'topics': [], 'publishedToDatabase': False, 'limits': ['保留作者、媒体原文和日期；不等同独立购买评价或已验证消费者身份', '原文报价、营养、工艺及功效声明未独立验证，不写为当前报价或功效结论', '原文可能含推广或转载；独立URL不等于独立证据，关联主题需按实际段落复核']} base.save(PUBLIC / (key + '.source.json'), row) return row def prior_discoveries(): old = PARENT / 'supplement-01' finals = {} for file in (old / 'public-sources').glob('*redirect-*.http.json'): row = json.loads(file.read_bytes()) if 'post.smzdm.com/' in row['finalUrl']: finals[file.stem.split('-redirect')[0]] = row['finalUrl'] result = {} for index in [2, 3, 4]: file = old / 'search' / ('query-' + str(index).zfill(2) + '.results.json') for row in json.loads(file.read_bytes())['results']: key = base.digest(row['url'].encode())[:24] url = finals.get(key) if not url or not re.fullmatch(r'https://post\.smzdm\.com/(?:talk/)?p/([a-z0-9]+)/?', url): continue if not url.endswith('/'): url += '/' # Discard general emergency stock and non-noodle alternatives at discovery. if re.search(r'方便面终结者|多囤5样|2千字', row['title']): continue result[url] = {**row, 'previousDiscoveryReceipt': str(file.relative_to(ROOT))} return result def publisher_search(index, query): url = 'https://www.foodaily.com/search/articles?' + urllib.parse.urlencode({'search': query}) key = 'publisher-' + str(index + 1).zfill(2) raw, receipt = base.fetch(url, key, SEARCH) root = html.fromstring(raw.decode('utf8')) if challenge(raw, receipt, root): raise ValueError('PUBLISHER_SEARCH_VERIFICATION') candidates = {} for anchor in root.xpath('//a[@href]'): link = anchor.get('href') title = re.sub(r'\s+', ' ', anchor.text_content()).strip() if not re.fullmatch(r'https://www\.foodaily\.com/articles/\d+', link) or not title: continue if not re.search(r'泡面|方便面|拉面|面饼|速食|面条|汤面|拌面', title): continue candidates[link] = {'url': link, 'title': title, 'query': query, 'receipt': key + '.http.json'} base.save(SEARCH / (key + '.results.json'), {'query': query, 'receipt': receipt, 'results': list(candidates.values())}) STATE['searches'].append({'kind': 'publisher-search', 'query': query, 'count': len(candidates), 'sha256': receipt['sha256']}) return candidates def general_search(index, topic, query): url = 'https://www.so.com/s?' + urllib.parse.urlencode({'q': query}) key = 'web-' + str(index + 1).zfill(2) raw, receipt = base.fetch(url, key, SEARCH) root = html.fromstring(raw.decode('utf8')) if challenge(raw, receipt, root): raise ValueError('WEB_SEARCH_VERIFICATION') result = [] for anchor in root.xpath('//a[@data-mdurl]'): link, title = anchor.get('data-mdurl'), re.sub(r'\s+', ' ', anchor.text_content()).strip() if not link or len(title) < 8: continue if re.fullmatch(r'https://(?:post\.smzdm\.com/(?:talk/)?p/[a-z0-9]+/?|sspai\.com/post/\d+)', link): result.append({'url': link if 'sspai.com' in link or link.endswith('/') else link + '/', 'title': title, 'topic': topic, 'query': query, 'receipt': key + '.http.json'}) base.save(SEARCH / (key + '.results.json'), {'query': query, 'receipt': receipt, 'results': result}) STATE['searches'].append({'kind': 'web-search', 'query': query, 'count': len(result), 'sha256': receipt['sha256']}) return result def main(): existing = existing_urls() candidates = prior_discoveries() candidates['https://sspai.com/post/73701'] = {'type': 'specific_original_review', 'topic': 'wellness'} for index, (topic, query) in enumerate(DISCOVERY): try: for row in general_search(index, topic, query): candidates.setdefault(row['url'], row) except Exception as error: STATE['rejected'].append({'query': query, 'reason': str(error)}) persist() time.sleep(1.3) for index, query in enumerate(QUERIES): try: candidates.update(publisher_search(index, query)) except Exception as error: STATE['rejected'].append({'query': query, 'reason': str(error)}) persist() time.sleep(1.1) # Aim at sparse topics first. A host's verification response stops it. candidates = {url: row for url, row in candidates.items() if url not in existing} STATE['candidateCount'] = len(candidates) print(json.dumps({'stage': 'discovery', 'newCandidates': len(candidates)}, ensure_ascii=False), flush=True) for index, (url, discovery) in enumerate(candidates.items()): try: result = collect(url, discovery) STATE['sources'].append(result) except Exception as error: STATE['rejected'].append({'url': url, 'reason': str(error)}) persist() print(json.dumps({'stage': 'body', 'processed': index + 1, 'target': len(candidates), 'sources': len(STATE['sources']), 'rejected': len(STATE['rejected'])}, ensure_ascii=False), flush=True) time.sleep(1.5) STATE['status'] = 'collection-complete-awaiting-semantic-review' STATE['finishedAt'] = base.now() STATE['uniqueSources'] = len(STATE['sources']) persist() base.save(WORK / 'review-candidates.json', [{'url': r['url'], 'platform': r['platform'], 'title': r['title'], 'date': r['publishedAtRaw'], 'body': r['body']} for r in STATE['sources']]) print(json.dumps({'status': STATE['status'], 'sources': len(STATE['sources']), 'hostStops': STATE['hostStops']}, ensure_ascii=False), flush=True) if __name__ == '__main__': main()