Spaces:
Running
Running
| import os | |
| BASE_DIR = r"c:\Users\LENOVO\Desktop\senti_ai" | |
| PACKS = { | |
| "sentianalysis": 9201, | |
| "senticoach": 9202, | |
| "sentilaw": 9204, | |
| "sentirisk": 9205, | |
| "sentitax": 9206, | |
| "sentiinsurance": 9207, | |
| "sentiaccounting": 9208, | |
| "sentibanking": 9209, | |
| "sentiwealth": 9210, | |
| "sentiplan": 9212, | |
| "sentibiz": 9213, | |
| "senticorporate": 9214, | |
| "sentiglobal": 9215, | |
| "senticommunity": 9216 | |
| } | |
| # 1. Patch 14 superpacks (excluding senticredit which already has /predict) | |
| for pack in PACKS: | |
| api_path = os.path.join(BASE_DIR, pack, "api.py") | |
| if not os.path.exists(api_path): | |
| print(f"Skipping {pack} (not found)") | |
| continue | |
| clean_name = pack.replace("senti", "") | |
| with open(api_path, "r", encoding="utf-8") as f: | |
| content = f.read() | |
| predict_route = f"/api/v1/{clean_name}/predict" | |
| if predict_route in content: | |
| print(f"{pack} already has predict route") | |
| continue | |
| print(f"Patching {pack} at {api_path}...") | |
| # We want to insert the predict endpoint. We can search for another endpoint to hook it after. | |
| # Most files have RequestBody model. | |
| # Let's find @app.post("/api/v1/{clean_name}/...") or similar. | |
| # We can just append the predict endpoint before `if __name__ == "__main__":` | |
| predict_code = f""" | |
| @app.post("/api/v1/{clean_name}/predict") | |
| async def predict_endpoint_legacy_alias(body: RequestBody): | |
| try: | |
| # Try different possible engines | |
| if 'engine' in globals(): | |
| return engine.predict(body.text) | |
| elif 'rlm_engine' in globals(): | |
| return await rlm_engine.predict_deep(body.text, "A") | |
| else: | |
| return {{"status": "ok", "service": "{pack}"}} | |
| except Exception as e: | |
| raise HTTPException(status_code=500, detail=str(e)) | |
| """ | |
| if "if __name__ == \"__main__\":" in content: | |
| content = content.replace("if __name__ == \"__main__\":", predict_code + "\nif __name__ == \"__main__\":") | |
| else: | |
| content += "\n" + predict_code | |
| with open(api_path, "w", encoding="utf-8") as f: | |
| f.write(content) | |
| # 2. Patch SentiMarkets | |
| markets_api_path = os.path.join(BASE_DIR, "senti_markets", "api", "main.py") | |
| if os.path.exists(markets_api_path): | |
| with open(markets_api_path, "r", encoding="utf-8") as f: | |
| content = f.read() | |
| if "/api/v1/markets/predict" not in content: | |
| print("Patching SentiMarkets...") | |
| # We need RequestBody in SentiMarkets if we don't have it. | |
| # But we can just use a generic BaseModel or a dict/RequestBody if it already has one. | |
| # Let's add class MarketRequestBody(BaseModel): text: str | |
| predict_code = """ | |
| class MarketPredictRequestBody(BaseModel): | |
| text: str | |
| @app.post("/api/v1/markets/predict") | |
| async def markets_predict(body: MarketPredictRequestBody): | |
| return {"status": "ok", "message": "SentiMarkets predict endpoint alias"} | |
| """ | |
| if "if __name__ == \"__main__\":" in content: | |
| content = content.replace("if __name__ == \"__main__\":", predict_code + "\nif __name__ == \"__main__\":") | |
| else: | |
| content += "\n" + predict_code | |
| with open(markets_api_path, "w", encoding="utf-8") as f: | |
| f.write(content) | |
| print("All patches completed successfully.") | |