import os import smtplib from email.mime.text import MIMEText from typing import List from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel import requests from huggingface_hub import InferenceClient # ====== KONFIGURACJA Z ENV ====== SERPAPI_KEY = os.getenv("SERPAPI_KEY") # klucz do SerpAPI HF_TOKEN = os.getenv("HF_TOKEN") # token Hugging Face HF_MODEL = os.getenv("HF_MODEL", "meta-llama/Meta-Llama-3-8B-Instruct") SMTP_HOST = os.getenv("SMTP_HOST", "smtp.gmail.com") SMTP_PORT = int(os.getenv("SMTP_PORT", "587")) SMTP_USER = os.getenv("SMTP_USER") SMTP_PASS = os.getenv("SMTP_PASS") FROM_EMAIL = os.getenv("FROM_EMAIL", SMTP_USER) # ====== FASTAPI ====== app = FastAPI() app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], ) client = InferenceClient(HF_MODEL, token=HF_TOKEN) # ====== MODELE DANYCH ====== class ScanRequest(BaseModel): query: str country: str | None = "pl" class Lead(BaseModel): title: str link: str snippet: str | None = None reason: str | None = None email_body: str | None = None class SendEmailRequest(BaseModel): to_email: str subject: str body: str # ====== POMOCNICZE ====== SOCIAL_DOMAINS = [ "facebook.com", "instagram.com", "panoramafirm.pl", "pkt.pl", "oferteo.pl", "olx.pl", "allegro.pl", "linkedin.com", "yellowpages", ] def is_own_website(url: str) -> bool: if not url: return False lower = url.lower() if any(d in lower for d in SOCIAL_DOMAINS): return False if "google.com/maps" in lower: return False return True def generate_email_text(company_name: str, context: str) -> str: prompt = f""" Jesteś specjalistą od sprzedaży stron internetowych dla lokalnych firm w Polsce. Napisz krótki, konkretny mail sprzedażowy po polsku do firmy: "{company_name}". Kontekst: {context} Założenia: - maksymalnie 2–3 krótkie akapity - bez lania wody - bez przesady z marketingiem - oferta: nowoczesna strona www + automatyzacja kontaktu (formularz, maile) - ton: profesjonalny, ale ludzki - bez 'Szanowni Państwo', tylko 'Dzień dobry' - zakończ pytaniem o możliwość krótkiej rozmowy lub wyceny """ resp = client.text_generation( prompt, max_new_tokens=220, temperature=0.6, do_sample=True, ) return resp.strip() def send_email(to_email: str, subject: str, body: str): msg = MIMEText(body, "plain", "utf-8") msg["Subject"] = subject msg["From"] = FROM_EMAIL msg["To"] = to_email with smtplib.SMTP(SMTP_HOST, SMTP_PORT) as server: server.starttls() server.login(SMTP_USER, SMTP_PASS) server.send_message(msg) # ====== ENDPOINT: SZUKANIE LEADÓW ====== @app.post("/scan", response_model=List[Lead]) def scan_leads(payload: ScanRequest): """ Szuka firm w Google (przez SerpAPI), filtruje te bez własnej strony www i generuje propozycję maila. """ if not SERPAPI_KEY: raise RuntimeError("Brak SERPAPI_KEY w env") params = { "engine": "google", "q": payload.query, "hl": "pl", "gl": payload.country or "pl", "api_key": SERPAPI_KEY, "num": 20, } r = requests.get("https://serpapi.com/search", params=params) data = r.json() results = data.get("organic_results", []) leads: List[Lead] = [] for item in results: title = item.get("title", "") link = item.get("link", "") snippet = item.get("snippet", "") # jeśli to własna strona → pomijamy if is_own_website(link): continue reason = "Brak własnej strony www (tylko katalog / social / listing)." email_body = generate_email_text(title, snippet or payload.query) leads.append( Lead( title=title, link=link, snippet=snippet, reason=reason, email_body=email_body, ) ) return leads # ====== ENDPOINT: WYSYŁANIE MAILA ====== @app.post("/send-email") def send_email_endpoint(payload: SendEmailRequest): if not (SMTP_USER and SMTP_PASS and FROM_EMAIL): raise RuntimeError("Brak konfiguracji SMTP w env") send_email(payload.to_email, payload.subject, payload.body) return {"status": "ok"}