#!/usr/bin/env python3 from __future__ import annotations import argparse, datetime as dt, hashlib, json, shutil, sqlite3, subprocess, uuid from pathlib import Path from typing import Any from zoneinfo import ZoneInfo from trendscout_google_trends_import import import_csv, inspect_csv HERE=Path(__file__).resolve().parent DEFAULT_DB=HERE.parent.parent.parent/"data/agents/trendscout/runtime/trendscout.sqlite3" DEFAULT_CONFIG=HERE/"weekly_intelligence_config.json" DEFAULT_VAULT=Path("/SHiREVault/SHIRE3D/TrendScout") NETWORK_FS={"cifs","nfs","nfs4","fuse.sshfs"} APPROVED={"APPROVED","GREEN","VERIFIED","COMMERCIAL_ALLOWED"} def now_utc(): return dt.datetime.now(dt.timezone.utc) def iso(v): return v.isoformat() def norm(v): return " ".join(str(v).strip().lower().split()) def read_json(path): value=json.loads(Path(path).read_text(encoding="utf-8")) if not isinstance(value,dict): raise ValueError(f"Expected object: {path}") return value def load_config(path): c=read_json(path) assert c.get("schemaVersion")==1 assert c.get("timezone")=="Australia/Perth" assert c["schedule"]["systemdCalendar"]=="Mon *-*-* 08:00:00 Australia/Perth" assert c["schedule"]["catchUpMissedRun"] is False assert c.get("trackedTerms") rules=c["rules"] unsafe=[k for k in ( "googleTrendsAloneProvesSales","searchInterestIsBuyerIntent", "automaticOpportunityCreation","automaticFileDownload","automaticPurchase", "automaticPrinting","automaticForgeHandoff","automaticListing", "automaticPublishing","automaticAdvertising","automaticSpending") if rules.get(k) is not False] if unsafe: raise ValueError("Unsafe rules: "+", ".join(unsafe)) assert rules.get("rayApprovalRequiredForExternalActions") is True return c def verify_vault(root, require_network=True): if require_network: p=subprocess.run(["findmnt","-n","-o","FSTYPE","-T","/SHiREVault"],capture_output=True,text=True,timeout=10) if p.returncode or p.stdout.strip() not in NETWORK_FS: raise RuntimeError("SHiREVault is not a confirmed network mount") root.mkdir(parents=True,exist_ok=True) token=uuid.uuid4().hex; test=root/f".write-test-{token}" try: test.write_text(token+"\n",encoding="utf-8") if test.read_text(encoding="utf-8").strip()!=token: raise RuntimeError("Vault write test failed") finally: test.unlink(missing_ok=True) def inbox(root): base=root/"weekly-inbox" paths={k:base/k for k in ("approved","processed","rejected")} for p in paths.values(): p.mkdir(parents=True,exist_ok=True) return paths def db(path, readonly=False): if readonly: c=sqlite3.connect(f"file:{Path(path).resolve()}?mode=ro",uri=True,timeout=30); c.execute("PRAGMA query_only=ON") else: c=sqlite3.connect(Path(path).resolve(),timeout=30); c.execute("PRAGMA foreign_keys=ON"); c.execute("PRAGMA busy_timeout=30000") c.row_factory=sqlite3.Row return c def destination(folder,name,label): stamp=now_utc().strftime("%Y%m%dT%H%M%SZ"); p=folder/f"{stamp}-{label}-{name}"; n=1 while p.exists(): p=folder/f"{stamp}-{label}-{n}-{name}"; n+=1 return p def duplicate(database,digest): c=db(database,True) try: return c.execute("SELECT 1 FROM source_records WHERE adapter_id='google-trends-manual' AND external_id=?",(f"google-trends-csv:{digest}",)).fetchone() is not None finally: c.close() def process_inbox(database,root,config): paths=inbox(root); out={"imported":[],"duplicates":[],"rejected":[]} for source in sorted(paths["approved"].glob("*.csv")): try: inspected=inspect_csv(source); digest=inspected["sha256"] if duplicate(database,digest): target=destination(paths["processed"],source.name,"duplicate"); shutil.move(source,target) out["duplicates"].append({"filename":source.name,"sha256":digest,"movedTo":str(target)}); continue result=import_csv(csv_path=source,database_path=database,vault_root=root,source_url=config["googleTrends"]["sourceUrl"],approved=True,require_network_vault=False) target=destination(paths["processed"],source.name,"imported"); shutil.move(source,target) out["imported"].append({**result,"processedPath":str(target)}) except Exception as exc: target=destination(paths["rejected"],source.name,"rejected") if source.exists(): shutil.move(source,target) Path(str(target)+".error.json").write_text(json.dumps({"errorType":type(exc).__name__,"error":str(exc),"rejectedAt":iso(now_utc())},indent=2)+"\n",encoding="utf-8") out["rejected"].append({"filename":source.name,"errorType":type(exc).__name__,"error":str(exc),"movedTo":str(target)}) return out def period(now,timezone): local=now.astimezone(ZoneInfo(timezone)); monday=local.date()-dt.timedelta(days=local.weekday()); sunday=monday+dt.timedelta(days=6); y,w,_=monday.isocalendar() return {"id":f"{y}-W{w:02d}","start":monday.isoformat(),"end":sunday.isoformat(),"local":local.isoformat()} def demand_rows(c,terms,limit): wanted={norm(x) for x in terms}; latest={} for row in c.execute("SELECT source_record_id,source_url,collected_at,payload_json,payload_sha256 FROM source_records WHERE adapter_id='google-trends-manual' AND is_demo=0 ORDER BY collected_at DESC"): try: payload=json.loads(row["payload_json"]) except Exception: continue for term,summary in (payload.get("summaries") or {}).items(): key=norm(term) if key not in wanted or key in latest or not isinstance(summary,dict): continue newest=float(summary.get("latest") or 0); recent=float(summary.get("recentAverage",summary.get("average")) or 0); direction=str(summary.get("direction") or "NO_DATA") bonus={"RISING":10,"FALLING":-10,"STABLE_OR_MIXED":0,"NO_DATA":-20}.get(direction,0) score=round(max(0,min(100,recent*.60+newest*.30+bonus)),2) latest[key]={"product":str(term),"normalisedProduct":key,"searchDemandScore":score,"signalClassification":"SEARCH_INTEREST_ONLY","direction":direction,"latestRelativeInterest":newest,"recentAverageRelativeInterest":recent,"averageRelativeInterest":summary.get("average"),"maximumRelativeInterest":summary.get("maximum"),"sampleCount":summary.get("sampleCount"),"sourceUrl":row["source_url"],"sourceRecordId":row["source_record_id"],"sourcePayloadSha256":row["payload_sha256"],"sourceArchiveSha256":payload.get("archiveSha256"),"collectedAt":row["collected_at"],"whyInDemand":f"Google Trends direction is {direction}; latest relative interest is {newest}. This is search-interest evidence, not proof of purchases or completed sales.","confirmedSales":False,"buyerIntent":False} ranked=sorted(latest.values(),key=lambda x:(x["searchDemandScore"],x["latestRelativeInterest"]),reverse=True)[:limit] for i,item in enumerate(ranked,1): item["rank"]=i return ranked def commercial_rows(c,limit): rows=c.execute(""" SELECT o.opportunity_id,o.product_name,o.category,o.status,o.opportunity_score,o.confidence,o.licence_status,o.independent_signals, COUNT(DISTINCT de.evidence_type) evidence_type_count, SUM(CASE WHEN de.confirmed_sale=1 THEN 1 ELSE 0 END) confirmed_sale_signals, SUM(CASE WHEN de.buyer_intent=1 THEN 1 ELSE 0 END) buyer_intent_signals FROM opportunities o LEFT JOIN demand_evidence de ON de.opportunity_id=o.opportunity_id AND de.is_demo=0 WHERE o.is_demo=0 AND UPPER(o.status) NOT IN ('REJECTED','BLOCKED','UNSAFE') GROUP BY o.opportunity_id ORDER BY confirmed_sale_signals DESC,buyer_intent_signals DESC,o.opportunity_score DESC LIMIT ?""",(limit,)).fetchall() result=[] for r in rows: sales=int(r["confirmed_sale_signals"] or 0); intent=int(r["buyer_intent_signals"] or 0); signals=int(r["independent_signals"] or 0) if signals<2 or (sales==0 and intent==0): continue f=c.execute("""SELECT fc.file_candidate_id,fc.creator_name,fc.source_url,fc.file_formats_json,fc.commercial_print_rights,fc.manual_review_required,lr.status review_status,lr.licence_name,lr.licence_url,lr.evidence_reference FROM file_candidates fc LEFT JOIN licence_reviews lr ON lr.file_candidate_id=fc.file_candidate_id WHERE fc.opportunity_id=? AND fc.commercial_print_rights=1 AND fc.manual_review_required=0 ORDER BY lr.reviewed_at DESC LIMIT 1""",(r["opportunity_id"],)).fetchone() review=str(f["review_status"] or "").upper() if f else ""; verified=bool(f and review in APPROVED) try: formats=json.loads(f["file_formats_json"] or "[]") if f else [] except Exception: formats=[] result.append({"opportunityId":r["opportunity_id"],"product":r["product_name"],"category":r["category"],"opportunityScore":r["opportunity_score"],"confidence":r["confidence"],"independentSignals":signals,"confirmedSaleSignals":sales,"buyerIntentSignals":intent,"evidenceTypeCount":int(r["evidence_type_count"] or 0),"licenceStatus":r["licence_status"],"verifiedCommercialFile":verified,"fileSourceUrl":f["source_url"] if f else None,"creatorOrStore":f["creator_name"] if f else None,"fileFormats":formats,"licenceName":f["licence_name"] if f else None,"licenceUrl":f["licence_url"] if f else None,"licenceEvidence":f["evidence_reference"] if f else None,"price":None,"printability":None,"estimatedPrintTime":None,"estimatedMaterial":None,"estimatedProductionCost":None,"estimatedSalePrice":None,"estimatedMargin":None,"competitionLevel":None,"recommendedSalesChannels":[],"shire3dSellingAngle":None,"missingCommercialFields":["price","printability","print time","material","production cost","sale price","margin","competition","sales channels","Shire3D selling angle"]}) for i,item in enumerate(result,1): item["rank"]=i return result def verified_files(c,limit): rows=c.execute("""SELECT fc.file_candidate_id,fc.opportunity_id,o.product_name,fc.creator_name,fc.source_url,fc.file_formats_json,fc.licence_status,lr.status review_status,lr.licence_name,lr.licence_url,lr.restrictions_json,lr.evidence_reference,lr.reviewed_at FROM file_candidates fc JOIN opportunities o ON o.opportunity_id=fc.opportunity_id LEFT JOIN licence_reviews lr ON lr.file_candidate_id=fc.file_candidate_id WHERE o.is_demo=0 AND fc.commercial_print_rights=1 AND fc.manual_review_required=0 ORDER BY lr.reviewed_at DESC LIMIT ?""",(limit,)).fetchall() out=[] for r in rows: if str(r["review_status"] or "").upper() not in APPROVED: continue try: formats=json.loads(r["file_formats_json"] or "[]") except Exception: formats=[] try: restrictions=json.loads(r["restrictions_json"] or "[]") except Exception: restrictions=[] out.append({"fileCandidateId":r["file_candidate_id"],"opportunityId":r["opportunity_id"],"product":r["product_name"],"creatorOrStore":r["creator_name"],"sourceUrl":r["source_url"],"fileFormats":formats,"licenceStatus":r["licence_status"],"licenceReviewStatus":r["review_status"],"licenceName":r["licence_name"],"licenceUrl":r["licence_url"],"commercialPhysicalPrintPermission":True,"digitalRedistributionPermission":False,"licenceRestrictions":restrictions,"licenceEvidenceReference":r["evidence_reference"],"licenceReviewedAt":r["reviewed_at"],"price":None,"printability":None,"estimatedCostAndMargin":None,"approvalRequiredBeforeDownloadOrPurchase":True}) return out def rejected(c,limit): return [dict(x) for x in c.execute("SELECT opportunity_id opportunityId,product_name product,category,status,opportunity_score opportunityScore,licence_status licenceStatus,updated_at updatedAt FROM opportunities WHERE is_demo=0 AND (UPPER(status) IN ('REJECTED','BLOCKED','UNSAFE') OR UPPER(licence_status) IN ('RED','REJECTED','BLOCKED','PROHIBITED')) ORDER BY updated_at DESC LIMIT ?",(limit,)).fetchall()] def coverage(c): sources=[dict(x) for x in c.execute("SELECT adapter_id,name,mode,enabled,health_status FROM source_integrations ORDER BY name")] evidence=[dict(x) for x in c.execute("SELECT evidence_type,COUNT(*) item_count,SUM(CASE WHEN buyer_intent=1 THEN 1 ELSE 0 END) buyer_intent_count,SUM(CASE WHEN confirmed_sale=1 THEN 1 ELSE 0 END) confirmed_sale_count FROM demand_evidence WHERE is_demo=0 GROUP BY evidence_type")] return {"configuredSources":sources,"enabledLiveSourceCount":sum(int(x["enabled"]) for x in sources),"evidenceByType":evidence,"searchInterestAvailable":any(x["evidence_type"]=="SEARCH_GROWTH" and int(x["item_count"])>0 for x in evidence),"buyerIntentAvailable":any(int(x["buyer_intent_count"] or 0)>0 for x in evidence),"confirmedSalesAvailable":any(int(x["confirmed_sale_count"] or 0)>0 for x in evidence)} def build_report(database,config,processing,now): p=period(now,config["timezone"]); c=db(database,True) try: cov=coverage(c); demand=demand_rows(c,config["trackedTerms"],config["limits"]["mostInDemand"]); commercial=commercial_rows(c,config["limits"]["commerciallyPromising"]); files=verified_files(c,config["limits"]["verifiedFiles"]); rejects=rejected(c,config["limits"]["rejected"]) finally: c.close() status="WAITING_FOR_GOOGLE_TRENDS_EVIDENCE" if not demand else "SEARCH_DEMAND_ONLY_MORE_SALES_EVIDENCE_REQUIRED" if not commercial else "DEMAND_VALIDATED_FILE_LICENCE_REVIEW_REQUIRED" if not files else "READY_FOR_RAY_REVIEW" verified_products={norm(x["product"]) for x in files} original=[{"product":x["product"],"demandRank":x["rank"],"searchDemandScore":x["searchDemandScore"],"reason":"Demand evidence exists, but no verified commercial STL/3MF/STEP file is stored.","forgeBriefStatus":"NOT_CREATED","rayApprovalRequired":True} for x in demand if norm(x["product"]) not in verified_products] actions=[] if not demand: actions.append({"action":"Place the newest original Google Trends Interest over time CSV in the approved inbox.","requiredFor":"Most-in-demand ranking","approval":"Placing it there is explicit Ray approval to import it."}) if demand and not cov["buyerIntentAvailable"]: actions.append({"action":"Connect or import a permitted buyer-intent or customer-enquiry source.","requiredFor":"Commercial validation","approval":"Required before enabling the source."}) if demand and not cov["confirmedSalesAvailable"]: actions.append({"action":"Connect a permitted transactional or Shire3D confirmed-sales source.","requiredFor":"Proven-seller ranking","approval":"Required before enabling the source."}) if demand and not files: actions.append({"action":"Research legitimate commercial STL, 3MF or STEP sources for the highest-ranked terms.","requiredFor":"Ready-to-print recommendations","approval":"Required before any download or purchase."}) actions.append({"action":"Review the weekly intelligence report.","requiredFor":"Any printing, Forge handoff, listing, publishing, advertising or spending","approval":"Explicit Ray approval is always required."}) return {"schemaVersion":1,"reportId":f"weekly_{p['id']}","reportingPeriod":p["id"],"reportingPeriodStart":p["start"],"reportingPeriodEnd":p["end"],"generatedAt":iso(now),"localGeneratedAt":p["local"],"timezone":config["timezone"],"schedule":config["schedule"],"status":status,"routineEmailCount":0,"demoRecordsExcluded":True,"trackedTerms":config["trackedTerms"],"googleTrendsSettings":config["googleTrends"],"approvedInboxProcessing":processing,"sourceCoverage":cov,"mostInDemand":demand,"commerciallyPromising":commercial,"verifiedCommercialFiles":files,"originalDesignOpportunities":original,"risingOpportunities":[x for x in demand if x["direction"]=="RISING"],"rejectedItems":rejects,"approvalActions":actions,"limitations":["Google Trends is relative search-interest evidence and does not prove sales.","Search popularity, social attention, listings, buyer intent and confirmed sales remain separate signals.","Commercial physical-print rights must be verified before launch.","Price, printability, cost, margin, competition, channel and selling-angle fields remain blank until supported evidence exists.","Weekly email delivery is not installed yet."],"noSafeLaunchRecommendation":not(bool(commercial) and bool(files)),"externalActionsPerformed":{"filesDownloaded":0,"filesPurchased":0,"printsStarted":0,"forgeHandoffs":0,"listingsCreated":0,"itemsPublished":0,"advertisingStarted":0,"moneySpent":0,"emailsSent":0}} def markdown(report): lines=[f"# TrendScout Weekly Intelligence — {report['reportingPeriod']}","",f"Status: {report['status']}",f"Generated: {report['localGeneratedAt']}","","## Most in demand"] if report["mostInDemand"]: for x in report["mostInDemand"]: lines.append(f"{x['rank']}. {x['product']} — score {x['searchDemandScore']} — {x['direction']} — search interest only") else: lines.append("No approved Google Trends evidence is available.") lines += ["","## Commercially promising"] if report["commerciallyPromising"]: for x in report["commerciallyPromising"]: lines.append(f"{x['rank']}. {x['product']} — sales signals {x['confirmedSaleSignals']} — buyer-intent signals {x['buyerIntentSignals']} — verified file {x['verifiedCommercialFile']}") else: lines.append("No opportunity yet has enough independent buyer-intent or confirmed-sales evidence.") lines += ["","## Verified commercial STL/3MF/STEP files"] if report["verifiedCommercialFiles"]: for x in report["verifiedCommercialFiles"]: lines.append(f"- {x['product']} — {x.get('creatorOrStore') or 'Unknown creator'} — {x['sourceUrl']} — {x.get('licenceName') or 'Licence name unavailable'}") else: lines.append("No commercially verified file is ready for recommendation.") lines += ["","## Original design opportunities"] for x in report["originalDesignOpportunities"]: lines.append(f"- {x['product']} — demand rank {x['demandRank']} — Forge brief not created; Ray approval required") lines += ["","## Approval actions"] for x in report["approvalActions"]: lines.append(f"- {x['action']} ({x['approval']})") lines += ["","## Limitations"]+[f"- {x}" for x in report["limitations"]] return "\n".join(lines)+"\n" def database_report_status(intelligence_status): # The repository's weekly_reports table uses its original lifecycle # contract. Detailed intelligence readiness stays inside report_json. if intelligence_status == "WAITING_FOR_GOOGLE_TRENDS_EVIDENCE": return "COLLECTING_DATA" return "DRAFT_READY" def persist(database,root,report): folder=root/"weekly-reports"/report["reportingPeriod"]; folder.mkdir(parents=True,exist_ok=True) database_status=database_report_status(report["status"]) c=db(database,False) try: existing=c.execute("SELECT report_id,created_at FROM weekly_reports WHERE reporting_period=? LIMIT 1",(report["reportingPeriod"],)).fetchone(); c.execute("BEGIN IMMEDIATE") rid=existing["report_id"] if existing else report["reportId"] report["reportId"]=rid json_text=json.dumps(report,indent=2,ensure_ascii=False,sort_keys=True)+"\n" if existing: c.execute("UPDATE weekly_reports SET status=?,routine_email_count=0,demo_records_excluded=1,report_json=?,updated_at=? WHERE report_id=?",(database_status,json_text,report["generatedAt"],rid)) else: c.execute("INSERT INTO weekly_reports(report_id,reporting_period,status,routine_email_count,demo_records_excluded,report_json,created_at,updated_at) VALUES(?,?,?,0,1,?,?,?)",(rid,report["reportingPeriod"],database_status,json_text,report["generatedAt"],report["generatedAt"])) c.execute("DELETE FROM weekly_report_items WHERE report_id=?",(rid,)) for rank,x in enumerate(report["commerciallyPromising"],1): c.execute("INSERT INTO weekly_report_items(report_id,opportunity_id,rank) VALUES(?,?,?)",(rid,x["opportunityId"],rank)) c.commit() except Exception: c.rollback(); raise finally: c.close() json_path=folder/"report.json"; json_path.write_text(json_text,encoding="utf-8") digest=hashlib.sha256(json_text.encode()).hexdigest(); (folder/"report.json.sha256").write_text(f"{digest} report.json\n",encoding="utf-8") md=markdown(report); (folder/"report.md").write_text(md,encoding="utf-8"); (folder/"report.md.sha256").write_text(f"{hashlib.sha256(md.encode()).hexdigest()} report.md\n",encoding="utf-8") return {"reportId":rid,"reportingPeriod":report["reportingPeriod"],"status":report["status"],"databaseStatus":database_status,"reportJsonPath":str(json_path),"reportMarkdownPath":str(folder/"report.md"),"reportSha256":digest} def run(database,config_path,root,require_network=True,now=None): config=load_config(config_path); verify_vault(root,require_network); processing=process_inbox(database,root,config); report=build_report(database,config,processing,now or now_utc()); saved=persist(database,root,report) return {"ok":True,"processing":processing,"report":saved,"mostInDemandCount":len(report["mostInDemand"]),"commerciallyPromisingCount":len(report["commerciallyPromising"]),"verifiedCommercialFileCount":len(report["verifiedCommercialFiles"]),"routineEmailsSent":0,"externalActionsPerformed":report["externalActionsPerformed"]} def check(database,config_path): config=load_config(config_path); c=db(database,True) try: integrity=c.execute("PRAGMA integrity_check").fetchone()[0]; source=c.execute("SELECT mode,enabled FROM source_integrations WHERE adapter_id='google-trends-manual'").fetchone() finally: c.close() assert integrity=="ok" and source and source["mode"]=="MANUAL_REVIEW_ONLY" and source["enabled"]==0 return {"ok":True,"timezone":config["timezone"],"calendar":config["schedule"]["systemdCalendar"],"trackedTerms":config["trackedTerms"],"databaseIntegrity":integrity,"googleTrendsMode":source["mode"],"googleTrendsEnabled":False,"weeklyEmailEnabled":False,"liveResearchEnabled":False} def main(): p=argparse.ArgumentParser(); p.add_argument("--database",type=Path,default=DEFAULT_DB); p.add_argument("--config",type=Path,default=DEFAULT_CONFIG); p.add_argument("--vault-root",type=Path,default=DEFAULT_VAULT); g=p.add_mutually_exclusive_group(required=True); g.add_argument("--run",action="store_true"); g.add_argument("--check",action="store_true"); g.add_argument("--dry-run",action="store_true"); a=p.parse_args() if a.run: result=run(a.database,a.config,a.vault_root,True) elif a.check: result=check(a.database,a.config) else: config=load_config(a.config); result={"ok":True,"dryRun":True,"report":build_report(a.database,config,{"imported":[],"duplicates":[],"rejected":[],"dryRun":True},now_utc())} print(json.dumps(result,indent=2,ensure_ascii=False)) if __name__=="__main__": main()