Spaces:
Running
Running
siddhm11
Claude Opus 5.5
feat(discovery): CLI for collection, baseline capture and comparison
245fdd3 | #!/usr/bin/env python3 | |
| """Collect/prepare Hugging Face candidates in local shadow storage; compare feeds. | |
| Use --watch for the long-running scheduler, or invoke `collect` from cron. | |
| One filesystem lock serializes collection/preparation on the same host. | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import asyncio | |
| import json | |
| from pathlib import Path | |
| import sys | |
| import time | |
| ROOT=Path(__file__).resolve().parents[1] | |
| sys.path.insert(0,str(ROOT)) | |
| from app.discovery_runtime import worker_lock | |
| async def cycle(store, prepare: bool, limit: int) -> dict: | |
| import httpx | |
| from app.discovery_worker import collect_once,prepare_once | |
| async with httpx.AsyncClient() as client: | |
| result={'collection':await collect_once(store,client)} | |
| if prepare:result['preparation']=await prepare_once(store,client,limit=limit) | |
| result['store']=await asyncio.to_thread(store.status,time.time()) | |
| from datetime import datetime, timezone | |
| result['checked_at']=datetime.now(timezone.utc).isoformat() | |
| return result | |
| def render_comparison(report: dict, output: Path) -> None: | |
| from jinja2 import Environment,FileSystemLoader,select_autoescape | |
| output.mkdir(parents=True,exist_ok=True) | |
| env=Environment(loader=FileSystemLoader(ROOT/'scripts'),autoescape=select_autoescape(['html'])) | |
| (output/'comparison.json').write_text(json.dumps(report,indent=2,allow_nan=False)) | |
| (output/'comparison.html').write_text(env.get_template('hf_comparison.html').render(report=report)) | |
| async def capture_baseline(seed_ids: list[str], output: Path) -> dict: | |
| """Run the existing feed builder for a scratch profile, without impressions.""" | |
| import tempfile | |
| import uuid | |
| from datetime import datetime, timezone | |
| from app import config, db, qdrant_svc, user_state | |
| from app.recommend import profiles | |
| from app.routers import recommendations | |
| from app.discovery_store import EMBEDDING_CONTRACT | |
| original=(config.DB_PATH,db.DB_PATH) | |
| uid='shadow-'+uuid.uuid4().hex | |
| try: | |
| with tempfile.TemporaryDirectory(prefix='researchit-baseline-') as temp: | |
| config.DB_PATH=db.DB_PATH=str(Path(temp)/'user.sqlite') | |
| await db.init_db() | |
| seed_ids=list(dict.fromkeys(seed_ids)) | |
| vecs=await asyncio.wait_for(qdrant_svc.get_paper_vectors(seed_ids),timeout=45) | |
| usable=[aid for aid in seed_ids if aid in vecs] | |
| if len(usable)<5:raise ValueError('Need at least five available seed vectors') | |
| state=await user_state.ensure_loaded(uid) | |
| for aid in usable: | |
| state.add_positive(aid) | |
| await db.log_interaction(uid,aid,'save') | |
| await profiles.update_on_save(uid,vecs[aid]) | |
| entry=await asyncio.wait_for(recommendations._build_feed(uid,state,uuid.uuid4().hex),timeout=90) | |
| if entry is None:raise ValueError('Baseline feed unavailable') | |
| papers,_=await recommendations._build_page(entry,set(usable)) | |
| papers=papers[:10] | |
| if not papers:raise ValueError('Baseline page empty') | |
| vectors=await asyncio.wait_for(qdrant_svc.get_paper_vectors([p['arxiv_id'] for p in papers]),timeout=45) | |
| if any(p['arxiv_id'] not in vectors for p in papers): | |
| raise ValueError('Incomplete baseline candidate vectors') | |
| result={'name':'Scratch profile from supplied seeds','embedding_contract':EMBEDDING_CONTRACT, | |
| 'captured_at':datetime.now(timezone.utc).isoformat(), | |
| 'seed_ids':usable,'seed_vectors':[vecs[aid].tolist() for aid in usable], | |
| 'papers':[{'arxiv_id':p['arxiv_id'],'title':p['title'], | |
| 'published_at':p.get('published',''),'vector':vectors[p['arxiv_id']].tolist()} | |
| for p in papers], | |
| 'excluded_ids':usable, | |
| 'provenance':'Actual feed builder, scratch seed profile. No delivered impressions.'} | |
| output.parent.mkdir(parents=True,exist_ok=True) | |
| output.write_text(json.dumps(result,allow_nan=False)) | |
| return {'captured':len(papers),'usable_seeds':len(usable)} | |
| finally: | |
| config.DB_PATH,db.DB_PATH=original | |
| user_state._cache.pop(uid,None) | |
| def synthetic_demo(output: Path) -> None: | |
| """Exercise collection-to-comparison using an explicitly synthetic source/model.""" | |
| import tempfile | |
| from datetime import datetime, timezone | |
| import httpx | |
| from app.discovery_store import DiscoveryStore, EMBEDDING_CONTRACT | |
| from app.discovery_worker import collect_once, prepare_once | |
| from app.discovery_shadow import compare_feed | |
| # Use a microsecond-exact clock across JSON and SQLite eligibility checks. | |
| now=datetime.now(timezone.utc).timestamp() | |
| date=datetime.fromtimestamp(now-86400,timezone.utc).isoformat() | |
| def vector(topic): | |
| a=[0.0]*1024;a[topic]=1.0;return a | |
| baseline={'name':'Synthetic two-interest reader','embedding_contract':EMBEDDING_CONTRACT, | |
| 'seed_ids':[f'seed-{i}' for i in range(10)], | |
| 'seed_vectors':[vector(i%2) for i in range(10)], | |
| 'papers':[{'arxiv_id':f'demo-baseline-{i}','title':f'Simulated interest {i%2+1}: baseline {i+1}', | |
| 'vector':vector(i%2),'published_at':date} for i in range(10)]} | |
| rows=[{'paper':{'id':f'2609.{i:05d}','title':f'Simulated fresh research {i+1}', | |
| 'summary':'Synthetic fixture, not a real paper.', 'publishedAt':date, | |
| 'upvotes':0}} for i in range(5)] | |
| async def run(store): | |
| async with httpx.AsyncClient(transport=httpx.MockTransport(lambda _:httpx.Response(200,json=rows))) as client: | |
| collected=await collect_once(store,client,now) | |
| prepared=await prepare_once(store,client,now=now,encoder=lambda p:vector(int(p['arxiv_id'][-1])%2)) | |
| report=compare_feed(baseline,store.ready(now),now) | |
| report.update(synthetic=True,store=store.status(now),execution={'collection':collected,'preparation':prepared}) | |
| render_comparison(report,output) | |
| with tempfile.TemporaryDirectory(prefix='researchit-shadow-demo-') as temp: | |
| asyncio.run(run(DiscoveryStore(Path(temp)/'source.sqlite'))) | |
| def main() -> int: | |
| p=argparse.ArgumentParser(description=__doc__) | |
| p.add_argument('--store',type=Path,default=ROOT/'data/discovery.sqlite') | |
| sub=p.add_subparsers(dest='command',required=True) | |
| collect=sub.add_parser('collect');collect.add_argument('--prepare',action='store_true') | |
| collect.add_argument('--limit',type=int,default=10) | |
| collect.add_argument('--report',type=Path,help='Write collection and readiness status as JSON') | |
| collect.add_argument('--watch',action='store_true');collect.add_argument('--interval',type=int,default=21600) | |
| sub.add_parser('status') | |
| capture=sub.add_parser('capture');capture.add_argument('--seeds',nargs='+',required=True) | |
| capture.add_argument('--output',type=Path,required=True) | |
| demo=sub.add_parser('demo');demo.add_argument('--output',type=Path,default=ROOT/'reports/recommendations/shadow-demo') | |
| compare=sub.add_parser('compare');compare.add_argument('--baseline',type=Path,required=True) | |
| compare.add_argument('--output',type=Path,default=ROOT/'reports/recommendations/shadow') | |
| args=p.parse_args() | |
| if args.command=='capture': | |
| try:print(json.dumps(asyncio.run(capture_baseline(args.seeds,args.output))));return 0 | |
| except Exception as exc: | |
| print(json.dumps({'status':'failed','reason':type(exc).__name__}));return 1 | |
| if args.command=='demo': | |
| synthetic_demo(args.output);print(str(args.output/'comparison.html'));return 0 | |
| from app.discovery_store import DiscoveryStore | |
| store=DiscoveryStore(args.store) | |
| if args.command=='status':print(json.dumps(store.status(time.time()),indent=2));return 0 | |
| if args.command=='compare': | |
| from app.discovery_shadow import compare_feed | |
| now=time.time() | |
| baseline=json.loads(args.baseline.read_text()) | |
| report=compare_feed(baseline,store.ready(now),now) | |
| report['store']=store.status(now) | |
| render_comparison(report,args.output) | |
| print(str(args.output/'comparison.html'));return 0 | |
| if args.interval<300 or not 1<=args.limit<=100:p.error('interval >=300 seconds; limit 1..100') | |
| # Hold for the worker's entire life; no racing cron or second --watch process. | |
| with worker_lock(args.store.with_suffix('.worker.lock')): | |
| while True: | |
| result=asyncio.run(cycle(store,args.prepare,args.limit)) | |
| if args.report: | |
| args.report.parent.mkdir(parents=True,exist_ok=True) | |
| temporary=args.report.with_suffix('.tmp') | |
| temporary.write_text(json.dumps(result,indent=2)) | |
| temporary.replace(args.report) | |
| print(json.dumps(result),flush=True) | |
| if not args.watch: | |
| return int(result['collection']['status']!='ok' or result.get('preparation',{}).get('failed',0)>0 or result.get('preparation',{}).get('blocked',False)) | |
| time.sleep(args.interval) | |
| if __name__=='__main__':raise SystemExit(main()) | |