Create app.py

app.py

@@ -0,0 +1,228 @@
+# ============================================================
+# analyzer_agent/app.py — Telegram Analyzer Agent (async)
+# Mamba + GGUF LLM + Pyrogram + FastAPI
+# ============================================================
+# Telegram
+#TG_API_ID=...
+#TG_API_HASH=...
+#TG_BOT_TOKEN=...        # بوت لديه صلاحية النشر في القناة
+#TG_CHANNEL=@my_channel  # أو -1001234567890
+
+# Image agent
+#HF_API_TOKEN=...        # إن كنت تستخدم HF
+#HF_MODEL=stabilityai/stable-diffusion-2
+#USE_LOCAL_DIFFUSERS=0   # 1 إن أردت استخدام diffusers محليًا وكنت على GPU
+
+import os
+import json
+import asyncio
+from datetime import datetime
+import logging
+
+from fastapi import FastAPI
+from pydantic import BaseModel
+from apscheduler.schedulers.asyncio import AsyncIOScheduler
+
+import torch
+from transformers import AutoTokenizer, AutoModelForCausalLM
+
+from pyrogram import Client
+
+# llama.cpp (GGUF)
+from llama_cpp import Llama
+from huggingface_hub import hf_hub_download
+
+
+# ---------------- Logging ----------------
+logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
+log = logging.getLogger("analyzer")
+
+
+# ---------------- Env & config ----------------
+TG_API_ID = int(os.getenv("TG_API_ID", "0"))
+TG_API_HASH = os.getenv("TG_API_HASH", "")
+TG_BOT_TOKEN = os.getenv("TG_BOT_TOKEN")
+TG_CHANNEL = os.getenv("TG_CHANNEL")
+LOG_PATH = os.getenv("ANALYZER_LOG", "analyzer_log.json")
+POSTS_LIMIT = int(os.getenv("ANALYZER_LIMIT", "80"))
+
+MAMBA_MODEL_PATH = os.getenv("MAMBA_MODEL_PATH", "state-spaces/mamba2-1.3b")
+
+# يتم تجاهل LLM_MODEL_PATH في حالة GGUF لكن نتركه للانسجام
+LLM_MODEL_PATH = os.getenv("LLM_MODEL_PATH", "unused_for_gguf")
+
+# ---------------- Load Mamba ----------------
+log.info("Loading Mamba model...")
+
+mamba_tok = AutoTokenizer.from_pretrained(MAMBA_MODEL_PATH)
+mamba_model = AutoModelForCausalLM.from_pretrained(
+    MAMBA_MODEL_PATH,
+    torch_dtype=torch.float16,
+    low_cpu_mem_usage=True
+)
+
+
+# ---------------- Load GGUF LLM (Zephyr 7B) ----------------
+log.info("Loading LLM interpreter (GGUF + llama.cpp)...")
+
+LLM_GGUF_REPO = "TheBloke/zephyr-7B-beta-GGUF"
+LLM_GGUF_FILE = "zephyr-7b-beta.Q6_K.gguf"
+
+LLM_LOCAL_PATH = os.getenv("LLM_GGUF_PATH", f"./{LLM_GGUF_FILE}")
+
+if not os.path.exists(LLM_LOCAL_PATH):
+    log.info("Downloading GGUF model from HuggingFace...")
+    LLM_LOCAL_PATH = hf_hub_download(
+        repo_id=LLM_GGUF_REPO,
+        filename=LLM_GGUF_FILE
+    )
+
+llm = Llama(
+    model_path=LLM_LOCAL_PATH,
+    n_ctx=4096,
+    n_threads=4,
+    n_gpu_layers=0  # إذا لديك GPU ضع قيمة أكبر
+)
+
+log.info("GGUF model loaded successfully.")
+
+
+# ---------------- Pyrogram Client ----------------
+tg_client = Client("analyzer_bot", api_id=TG_API_ID, api_hash=TG_API_HASH, bot_token=TG_BOT_TOKEN)
+
+
+# ---------------- FastAPI ----------------
+app = FastAPI(title="Analyzer Agent")
+
+
+# ---------------- Helpers ----------------
+def save_log(entry):
+    logs = []
+    if os.path.exists(LOG_PATH):
+        try:
+            with open(LOG_PATH, "r", encoding="utf-8") as f:
+                logs = json.load(f)
+        except Exception:
+            logs = []
+    logs.insert(0, entry)
+    with open(LOG_PATH, "w", encoding="utf-8") as f:
+        json.dump(logs, f, ensure_ascii=False, indent=2)
+
+
+def encode_stats_for_mamba(posts):
+    seq = []
+    for p in posts:
+        seq.append(f"[{p['id']}: VW={p['views']}, FW={p['forwards']}, RC={p['reactions']}]")
+    return " ".join(seq)
+
+
+def run_mamba(text):
+    inp = mamba_tok(text, return_tensors="pt")
+    with torch.no_grad():
+        out = mamba_model.generate(**inp, max_new_tokens=64, do_sample=False)
+    return mamba_tok.decode(out[0], skip_special_tokens=True)
+
+
+def interpret_with_llm(mamba_output):
+    prompt = (
+        "هذه نتائج تحليل إحصائي لقناة تلغرام:\n"
+        f"{mamba_output}\n\n"
+        "حلل الأداء واستخرج:\n"
+        "- نقاط القوة\n"
+        "- نقاط الضعف\n"
+        "- أفضل أوقات النشر المتوقعة\n"
+        "- نوع المحتوى الذي يرفع الوصول\n"
+        "- استراتيجيات لزيادة الاشتراكات والتفاعل\n"
+        "اكتب التحليل بالعربية وبشكل مرتب ومختصر."
+    )
+
+    res = llm(
+        prompt,
+        max_tokens=250,
+        temperature=0.3,
+        top_p=0.95
+    )
+
+    return res["choices"][0]["text"].strip()
+
+
+# ---------------- Fetch Telegram Stats ----------------
+async def fetch_telegram_stats(limit=POSTS_LIMIT):
+    posts = []
+    async with tg_client:
+        async for msg in tg_client.get_chat_history(TG_CHANNEL, limit=limit):
+            if msg is None:
+                continue
+            views = getattr(msg, "views", 0) or 0
+            forwards = getattr(msg, "forwards", 0) or 0
+
+            reactions = 0
+            if getattr(msg, "reactions", None):
+                try:
+                    reactions = sum([r.count for r in msg.reactions.reactions])
+                except Exception:
+                    reactions = 0
+
+            posts.append({
+                "id": msg.message_id if hasattr(msg, "message_id") else msg.id,
+                "date": msg.date.isoformat() if getattr(msg, "date", None) else None,
+                "views": views,
+                "forwards": forwards,
+                "reactions": reactions
+            })
+    return posts
+
+
+# ---------------- Main pipeline ----------------
+async def daily_job():
+    log.info("Running daily analysis job...")
+    posts = await fetch_telegram_stats()
+
+    if not posts:
+        log.warning("No posts found for analysis.")
+        entry = {"time": datetime.utcnow().isoformat(), "error": "no_posts"}
+        save_log(entry)
+        return entry
+
+    stats_text = encode_stats_for_mamba(posts)
+    mamba_out = run_mamba(stats_text)
+    interpretation = interpret_with_llm(mamba_out)
+
+    entry = {
+        "time": datetime.utcnow().isoformat(),
+            "posts_count": len(posts),
+            "stats_text": stats_text,
+            "mamba_output": mamba_out,
+            "advice": interpretation
+        }
+
+        save_log(entry)
+        log.info("Analysis saved.")
+        return entry
+
+
+# ---------------- API endpoints ----------------
+@app.get("/run_once")
+async def run_once():
+    return await daily_job()
+
+
+@app.get("/logs")
+def get_logs():
+    if os.path.exists(LOG_PATH):
+        with open(LOG_PATH, "r", encoding="utf-8") as f:
+            return json.load(f)
+    return []
+
+
+# ---------------- Scheduler ----------------
+scheduler = AsyncIOScheduler()
+scheduler.add_job(lambda: asyncio.create_task(daily_job()), "cron", hour=0, minute=5)
+scheduler.start()
+
+
+# ---------------- Main ----------------
+if __name__ == "__main__":
+    import uvicorn
+    log.info("Starting Analyzer Agent...")
+    uvicorn.run("analyzer_agent.app:app", host="0.0.0.0", port=int(os.getenv("PORT", "7861")), reload=False)