from dotenv import load_dotenv from pathlib import Path ROOT_DIR = Path(__file__).parent load_dotenv(ROOT_DIR / ".env") import os import logging import uuid import asyncio import secrets from datetime import datetime, timezone, timedelta from typing import List, Optional from zoneinfo import ZoneInfo import bcrypt import jwt import httpx import resend from apscheduler.schedulers.asyncio import AsyncIOScheduler from apscheduler.triggers.cron import CronTrigger from bson import ObjectId from fastapi import FastAPI, APIRouter, HTTPException, Request, Response, Depends, Query from fastapi.responses import JSONResponse, HTMLResponse from starlette.middleware.cors import CORSMiddleware from motor.motor_asyncio import AsyncIOMotorClient from pydantic import BaseModel, Field, EmailStr, ConfigDict # ---------- LLM Integration (Gemini 3 Pro via Emergent) ---------- from emergentintegrations.llm.chat import LlmChat, UserMessage # ============= Config ============= MONGO_URL = os.environ["MONGO_URL"] DB_NAME = os.environ["DB_NAME"] JWT_SECRET = os.environ["JWT_SECRET"] JWT_ALGORITHM = "HS256" ACCESS_TOKEN_EXPIRE_MINUTES = 60 * 24 # 1 day for usability REFRESH_TOKEN_EXPIRE_DAYS = 7 ADMIN_EMAIL = os.environ.get("ADMIN_EMAIL", "admin@example.com") ADMIN_PASSWORD = os.environ.get("ADMIN_PASSWORD", "admin123") EMERGENT_LLM_KEY = os.environ.get("EMERGENT_LLM_KEY", "") FRONTEND_URL = os.environ.get("FRONTEND_URL", "http://localhost:3000") # Newsletter / Resend RESEND_API_KEY = os.environ.get("RESEND_API_KEY", "") SENDER_EMAIL = os.environ.get("SENDER_EMAIL", "onboarding@resend.dev") SENDER_NAME = os.environ.get("SENDER_NAME", "CryptoEditorial") NEWSLETTER_HOUR_JAKARTA = int(os.environ.get("NEWSLETTER_HOUR_JAKARTA", "8")) JAKARTA_TZ = ZoneInfo("Asia/Jakarta") if RESEND_API_KEY: resend.api_key = RESEND_API_KEY # Cookie config — for HTTPS production set COOKIE_SECURE=true & COOKIE_SAMESITE=none # For local HTTP development (e.g. Laragon) set COOKIE_SECURE=false & COOKIE_SAMESITE=lax COOKIE_SECURE = os.environ.get("COOKIE_SECURE", "true").lower() == "true" COOKIE_SAMESITE = os.environ.get("COOKIE_SAMESITE", "none") COINGECKO_BASE = "https://api.coingecko.com/api/v3" COINGECKO_API_KEY = os.environ.get("COINGECKO_API_KEY", "") # ============= Mongo ============= client = AsyncIOMotorClient(MONGO_URL) db = client[DB_NAME] # ============= App & Router ============= app = FastAPI(title="CryptoEditorial API") api_router = APIRouter(prefix="/api") # Configure logging logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s") logger = logging.getLogger(__name__) # ============= Helpers - password & jwt ============= def hash_password(password: str) -> str: salt = bcrypt.gensalt() return bcrypt.hashpw(password.encode("utf-8"), salt).decode("utf-8") def verify_password(plain: str, hashed: str) -> bool: try: return bcrypt.checkpw(plain.encode("utf-8"), hashed.encode("utf-8")) except Exception: return False def create_access_token(user_id: str, email: str) -> str: payload = { "sub": user_id, "email": email, "type": "access", "exp": datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES), } return jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM) def create_refresh_token(user_id: str) -> str: payload = { "sub": user_id, "type": "refresh", "exp": datetime.now(timezone.utc) + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS), } return jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM) def set_auth_cookies(response: Response, access_token: str, refresh_token: str): response.set_cookie( key="access_token", value=access_token, httponly=True, secure=COOKIE_SECURE, samesite=COOKIE_SAMESITE, max_age=ACCESS_TOKEN_EXPIRE_MINUTES * 60, path="/", ) response.set_cookie( key="refresh_token", value=refresh_token, httponly=True, secure=COOKIE_SECURE, samesite=COOKIE_SAMESITE, max_age=REFRESH_TOKEN_EXPIRE_DAYS * 24 * 3600, path="/", ) def clear_auth_cookies(response: Response): response.delete_cookie("access_token", path="/") response.delete_cookie("refresh_token", path="/") # ============= Auth dependency ============= async def get_current_user(request: Request) -> dict: token = request.cookies.get("access_token") if not token: auth_header = request.headers.get("Authorization", "") if auth_header.startswith("Bearer "): token = auth_header[7:] if not token: raise HTTPException(status_code=401, detail="Not authenticated") try: payload = jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGORITHM]) if payload.get("type") != "access": raise HTTPException(status_code=401, detail="Invalid token type") user_id = payload.get("sub") user = await db.users.find_one({"id": user_id}, {"_id": 0, "password_hash": 0}) if not user: raise HTTPException(status_code=401, detail="User not found") return user except jwt.ExpiredSignatureError: raise HTTPException(status_code=401, detail="Token expired") except jwt.InvalidTokenError: raise HTTPException(status_code=401, detail="Invalid token") # ============= Models ============= class RegisterRequest(BaseModel): email: EmailStr password: str = Field(min_length=6) name: str = Field(min_length=1) class LoginRequest(BaseModel): email: EmailStr password: str class UserPublic(BaseModel): id: str email: str name: str role: str created_at: datetime class HoldingCreate(BaseModel): coin_id: str # coingecko id e.g. "bitcoin" symbol: str name: str quantity: float buy_price_usd: float note: Optional[str] = "" class Holding(BaseModel): id: str coin_id: str symbol: str name: str quantity: float buy_price_usd: float note: str created_at: datetime class AnalyzeRequest(BaseModel): coin_id: str horizon: Optional[str] = "1m" # 1w, 1m, 3m # ============= Brute force protection ============= async def check_brute_force(identifier: str) -> None: rec = await db.login_attempts.find_one({"identifier": identifier}) if rec and rec.get("count", 0) >= 5: locked_until = rec.get("locked_until") if locked_until and datetime.now(timezone.utc) < locked_until: raise HTTPException(status_code=429, detail="Too many failed attempts. Try again in 15 minutes.") async def record_failed_login(identifier: str) -> None: rec = await db.login_attempts.find_one({"identifier": identifier}) count = (rec.get("count", 0) if rec else 0) + 1 update = {"count": count, "identifier": identifier} if count >= 5: update["locked_until"] = datetime.now(timezone.utc) + timedelta(minutes=15) await db.login_attempts.update_one({"identifier": identifier}, {"$set": update}, upsert=True) async def clear_failed_logins(identifier: str) -> None: await db.login_attempts.delete_one({"identifier": identifier}) # ============= AUTH ROUTES ============= @api_router.post("/auth/register") async def register(payload: RegisterRequest, response: Response): email = payload.email.lower() existing = await db.users.find_one({"email": email}) if existing: raise HTTPException(status_code=400, detail="Email already registered") user_doc = { "id": str(uuid.uuid4()), "email": email, "name": payload.name.strip(), "role": "user", "password_hash": hash_password(payload.password), "created_at": datetime.now(timezone.utc).isoformat(), } await db.users.insert_one(user_doc) access = create_access_token(user_doc["id"], email) refresh = create_refresh_token(user_doc["id"]) set_auth_cookies(response, access, refresh) return { "id": user_doc["id"], "email": email, "name": user_doc["name"], "role": user_doc["role"], "created_at": user_doc["created_at"], } @api_router.post("/auth/login") async def login(payload: LoginRequest, request: Request, response: Response): email = payload.email.lower() # Behind k8s ingress, request.client.host rotates between pod IPs. # Use first X-Forwarded-For IP (real client) or fall back. Fall back to email-only when no real IP. xff = request.headers.get("x-forwarded-for", "") real_ip = xff.split(",")[0].strip() if xff else (request.client.host if request.client else "unknown") identifier = f"{real_ip}:{email}" await check_brute_force(identifier) user = await db.users.find_one({"email": email}) if not user or not verify_password(payload.password, user["password_hash"]): await record_failed_login(identifier) raise HTTPException(status_code=401, detail="Invalid email or password") await clear_failed_logins(identifier) access = create_access_token(user["id"], email) refresh = create_refresh_token(user["id"]) set_auth_cookies(response, access, refresh) return { "id": user["id"], "email": user["email"], "name": user["name"], "role": user["role"], "created_at": user["created_at"], } @api_router.post("/auth/logout") async def logout(response: Response, _user: dict = Depends(get_current_user)): clear_auth_cookies(response) return {"ok": True} @api_router.get("/auth/me") async def me(user: dict = Depends(get_current_user)): return user @api_router.post("/auth/refresh") async def refresh_token(request: Request, response: Response): token = request.cookies.get("refresh_token") if not token: raise HTTPException(status_code=401, detail="No refresh token") try: payload = jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGORITHM]) if payload.get("type") != "refresh": raise HTTPException(status_code=401, detail="Invalid token type") user = await db.users.find_one({"id": payload["sub"]}, {"_id": 0}) if not user: raise HTTPException(status_code=401, detail="User not found") access = create_access_token(user["id"], user["email"]) new_refresh = create_refresh_token(user["id"]) set_auth_cookies(response, access, new_refresh) return {"ok": True} except jwt.ExpiredSignatureError: raise HTTPException(status_code=401, detail="Refresh token expired") except jwt.InvalidTokenError: raise HTTPException(status_code=401, detail="Invalid refresh token") # ============= COINGECKO PROXY ============= # Cache layer (avoid CoinGecko free-tier rate limits) _cache: dict = {} async def cached_fetch(url: str, ttl_seconds: int = 60, params: Optional[dict] = None) -> dict | list: key = url + "?" + "&".join(f"{k}={v}" for k, v in sorted((params or {}).items())) now = datetime.now(timezone.utc).timestamp() cached = _cache.get(key) if cached and cached["expires"] > now: return cached["data"] async with httpx.AsyncClient(timeout=20.0) as http: cg_headers = {"Accept": "application/json"} if COINGECKO_API_KEY: cg_headers["x-cg-demo-api-key"] = COINGECKO_API_KEY # Try once, retry once after a short delay on rate limit for attempt in range(2): r = await http.get(url, params=params, headers=cg_headers) if r.status_code == 200: data = r.json() if (isinstance(data, dict) and isinstance(data.get("status"), dict) and data["status"].get("error_code") == 429): # Embedded rate-limit error if attempt == 0: await asyncio.sleep(2) continue if cached: return cached["data"] raise HTTPException(status_code=503, detail="CoinGecko rate limit reached. Try again in a minute.") _cache[key] = {"data": data, "expires": now + ttl_seconds} return data if r.status_code == 429: if attempt == 0: await asyncio.sleep(2) continue if cached: return cached["data"] raise HTTPException(status_code=503, detail="CoinGecko rate limit reached. Try again in a minute.") # Other error if cached: return cached["data"] raise HTTPException(status_code=502, detail=f"CoinGecko error: {r.status_code}") # Should not reach here if cached: return cached["data"] raise HTTPException(status_code=503, detail="CoinGecko unavailable.") @api_router.get("/crypto/markets") async def crypto_markets( vs_currency: str = "usd", per_page: int = 50, page: int = 1, order: str = "market_cap_desc", ): """Top coins by market cap with 24h, 7d, 30d % changes and 7d sparkline.""" data = await cached_fetch( f"{COINGECKO_BASE}/coins/markets", ttl_seconds=45, params={ "vs_currency": vs_currency, "order": order, "per_page": per_page, "page": page, "sparkline": "true", "price_change_percentage": "1h,24h,7d,30d", }, ) return data @api_router.get("/crypto/top-gainers") async def top_gainers(period: str = "30d", vs_currency: str = "usd", limit: int = 10): """Top gainers in the last 30 days. CoinGecko returns market list, sort by selected period.""" if period not in ("24h", "7d", "30d"): period = "30d" raw = await cached_fetch( f"{COINGECKO_BASE}/coins/markets", ttl_seconds=60, params={ "vs_currency": vs_currency, "order": "market_cap_desc", "per_page": 250, "page": 1, "sparkline": "true", "price_change_percentage": "24h,7d,30d", }, ) field = { "24h": "price_change_percentage_24h_in_currency", "7d": "price_change_percentage_7d_in_currency", "30d": "price_change_percentage_30d_in_currency", }[period] sorted_coins = sorted( [c for c in raw if c.get(field) is not None], key=lambda c: c[field], reverse=True, ) return sorted_coins[:limit] @api_router.get("/crypto/coin/{coin_id}") async def coin_detail(coin_id: str): data = await cached_fetch( f"{COINGECKO_BASE}/coins/{coin_id}", ttl_seconds=180, params={ "localization": "false", "tickers": "false", "market_data": "true", "community_data": "false", "developer_data": "false", "sparkline": "true", }, ) # Trim down response - frontend doesn't need everything md = data.get("market_data", {}) or {} return { "id": data.get("id"), "symbol": data.get("symbol"), "name": data.get("name"), "image": (data.get("image") or {}).get("large"), "description": (data.get("description") or {}).get("en", "")[:1000], "categories": data.get("categories", []), "market_cap_rank": data.get("market_cap_rank"), "current_price": md.get("current_price", {}), "market_cap": md.get("market_cap", {}), "total_volume": md.get("total_volume", {}), "high_24h": md.get("high_24h", {}), "low_24h": md.get("low_24h", {}), "price_change_percentage_24h": md.get("price_change_percentage_24h"), "price_change_percentage_7d": md.get("price_change_percentage_7d"), "price_change_percentage_30d": md.get("price_change_percentage_30d"), "ath": md.get("ath", {}), "atl": md.get("atl", {}), "circulating_supply": md.get("circulating_supply"), "total_supply": md.get("total_supply"), "sparkline_7d": (md.get("sparkline_7d") or {}).get("price", []), } @api_router.get("/crypto/chart/{coin_id}") async def coin_chart(coin_id: str, days: int = 30, vs_currency: str = "usd"): params = {"vs_currency": vs_currency, "days": days} if days >= 90: params["interval"] = "daily" data = await cached_fetch( f"{COINGECKO_BASE}/coins/{coin_id}/market_chart", ttl_seconds=300, params=params, ) # Convert prices [[ts, price], ...] -> [{t, p}] prices = [{"t": p[0], "p": p[1]} for p in (data.get("prices") or [])] return {"prices": prices} @api_router.get("/crypto/search") async def crypto_search(query: str = Query(..., min_length=1)): data = await cached_fetch(f"{COINGECKO_BASE}/search", ttl_seconds=300, params={"query": query}) return {"coins": data.get("coins", [])[:15]} @api_router.get("/crypto/exchange-rate") async def exchange_rate(): """Returns USD <-> IDR rate for client-side conversion fallback (CoinGecko exchange_rates).""" data = await cached_fetch(f"{COINGECKO_BASE}/exchange_rates", ttl_seconds=300) rates = data.get("rates", {}) or {} return { "usd": rates.get("usd", {}).get("value"), "idr": rates.get("idr", {}).get("value"), } # ============= AI ANALYZER ============= @api_router.post("/ai/analyze") async def ai_analyze(req: AnalyzeRequest): """Generate buy/sell signal analysis for a given coin using Gemini 3 Pro.""" if not EMERGENT_LLM_KEY: raise HTTPException(status_code=500, detail="LLM key not configured") # Pull latest coin info try: coin = await coin_detail(req.coin_id) except HTTPException as e: raise HTTPException(status_code=400, detail=f"Failed to load coin info: {e.detail}") horizon_label = {"1w": "next 1 week", "1m": "next 1 month", "3m": "next 3 months"}.get(req.horizon, "next 1 month") prompt = f"""Anda adalah analis pasar crypto profesional yang memberikan analisis edukasi (BUKAN nasihat finansial). DATA COIN: - Nama: {coin.get('name')} ({(coin.get('symbol') or '').upper()}) - Rank Market Cap: #{coin.get('market_cap_rank')} - Harga (USD): ${(coin.get('current_price') or {}).get('usd')} - Perubahan 24 jam: {coin.get('price_change_percentage_24h')}% - Perubahan 7 hari: {coin.get('price_change_percentage_7d')}% - Perubahan 30 hari: {coin.get('price_change_percentage_30d')}% - ATH (USD): ${(coin.get('ath') or {}).get('usd')} - ATL (USD): ${(coin.get('atl') or {}).get('usd')} - Market Cap (USD): ${(coin.get('market_cap') or {}).get('usd')} - Volume 24h (USD): ${(coin.get('total_volume') or {}).get('usd')} - Kategori: {', '.join((coin.get('categories') or [])[:5])} HORIZON: {horizon_label} Berikan jawaban dalam Bahasa Indonesia dengan struktur JSON ketat berikut (HANYA JSON, tanpa text lain, tanpa code fences): {{ "signal": "BUY" | "HOLD" | "SELL", "confidence": 0-100, "summary": "Ringkasan 2-3 kalimat tentang kondisi terkini coin ini.", "buy_zone": "Rentang harga ideal untuk membeli (USD), contoh: $X - $Y", "sell_zone": "Rentang harga target jual (USD), contoh: $X - $Y", "stop_loss": "Level stop loss yang disarankan (USD)", "key_factors": ["3-5 faktor utama yang mendukung sinyal ini"], "risks": ["2-4 risiko utama yang perlu diwaspadai"], "timing_advice": "Saran kapan membeli dan menjual untuk horizon {horizon_label}" }}""" chat = LlmChat( api_key=EMERGENT_LLM_KEY, session_id=f"analyze-{req.coin_id}-{uuid.uuid4().hex[:8]}", system_message=( "You are a professional crypto market analyst providing educational analysis. " "You always respond with valid JSON only, no markdown, no code fences." ), ).with_model("gemini", "gemini-3.1-pro-preview") try: response_text = await chat.send_message(UserMessage(text=prompt)) except Exception as e: logger.exception("LLM call failed") raise HTTPException(status_code=502, detail=f"AI service error: {str(e)[:200]}") # Parse JSON robustly import json as _json text = (response_text or "").strip() if text.startswith("```"): text = text.strip("`") # remove leading "json" if text.lower().startswith("json"): text = text[4:] text = text.strip() # Find first { ... last } start = text.find("{") end = text.rfind("}") parsed = None if start != -1 and end != -1: try: parsed = _json.loads(text[start : end + 1]) except Exception: parsed = None if not parsed: # Fallback: return raw text in summary parsed = { "signal": "HOLD", "confidence": 50, "summary": text[:600] or "Tidak ada analisis dihasilkan.", "buy_zone": "-", "sell_zone": "-", "stop_loss": "-", "key_factors": [], "risks": [], "timing_advice": "-", } return { "coin": { "id": coin.get("id"), "name": coin.get("name"), "symbol": coin.get("symbol"), "image": coin.get("image"), "current_price_usd": (coin.get("current_price") or {}).get("usd"), "current_price_idr": (coin.get("current_price") or {}).get("idr"), }, "horizon": req.horizon, "analysis": parsed, "disclaimer": "Analisis ini dihasilkan AI untuk tujuan edukasi. BUKAN nasihat finansial. Selalu lakukan riset Anda sendiri (DYOR).", "generated_at": datetime.now(timezone.utc).isoformat(), } # ============= PORTFOLIO ============= @api_router.get("/portfolio/holdings") async def list_holdings(user: dict = Depends(get_current_user)): holdings = await db.holdings.find({"user_id": user["id"]}, {"_id": 0}).to_list(500) # Enrich with current price if not holdings: return {"holdings": [], "summary": {"total_invested_usd": 0, "total_value_usd": 0, "pnl_usd": 0, "pnl_pct": 0}} coin_ids = list({h["coin_id"] for h in holdings}) try: prices = await cached_fetch( f"{COINGECKO_BASE}/simple/price", ttl_seconds=30, params={"ids": ",".join(coin_ids), "vs_currencies": "usd,idr", "include_24hr_change": "true"}, ) except Exception: prices = {} total_invested = 0.0 total_value = 0.0 out = [] for h in holdings: cp = (prices.get(h["coin_id"]) or {}).get("usd", 0) or 0 cp_idr = (prices.get(h["coin_id"]) or {}).get("idr", 0) or 0 change_24h = (prices.get(h["coin_id"]) or {}).get("usd_24h_change", 0) or 0 invested = h["quantity"] * h["buy_price_usd"] value = h["quantity"] * cp pnl = value - invested pnl_pct = (pnl / invested * 100) if invested > 0 else 0 total_invested += invested total_value += value out.append( { **h, "current_price_usd": cp, "current_price_idr": cp_idr, "current_value_usd": value, "invested_usd": invested, "pnl_usd": pnl, "pnl_pct": pnl_pct, "change_24h": change_24h, } ) summary = { "total_invested_usd": total_invested, "total_value_usd": total_value, "pnl_usd": total_value - total_invested, "pnl_pct": ((total_value - total_invested) / total_invested * 100) if total_invested > 0 else 0, } return {"holdings": out, "summary": summary} @api_router.post("/portfolio/holdings") async def create_holding(payload: HoldingCreate, user: dict = Depends(get_current_user)): doc = { "id": str(uuid.uuid4()), "user_id": user["id"], "coin_id": payload.coin_id.lower(), "symbol": payload.symbol.upper(), "name": payload.name, "quantity": payload.quantity, "buy_price_usd": payload.buy_price_usd, "note": payload.note or "", "created_at": datetime.now(timezone.utc).isoformat(), } await db.holdings.insert_one(doc) return {k: v for k, v in doc.items() if k not in ("user_id", "_id")} @api_router.delete("/portfolio/holdings/{holding_id}") async def delete_holding(holding_id: str, user: dict = Depends(get_current_user)): result = await db.holdings.delete_one({"id": holding_id, "user_id": user["id"]}) if result.deleted_count == 0: raise HTTPException(status_code=404, detail="Holding not found") return {"ok": True} # ============= NEWSLETTER ============= class SubscribeRequest(BaseModel): email: EmailStr def _format_usd(v): if v is None: return "—" try: v = float(v) except Exception: return "—" if v >= 1: return "$" + f"{v:,.2f}" return "$" + f"{v:.6f}".rstrip("0").rstrip(".") def _format_idr(v): if v is None: return "—" try: v = float(v) except Exception: return "—" return "Rp " + f"{v:,.0f}".replace(",", ".") def _format_pct(v): if v is None: return "—" try: v = float(v) except Exception: return "—" sign = "+" if v >= 0 else "" return f"{sign}{v:.2f}%" async def build_newsletter_html() -> dict: """Build today's newsletter HTML. Returns {subject, html}.""" # 1. Top 5 gainers (24h) try: gainers = await top_gainers(period="24h", vs_currency="usd", limit=5) except Exception: logger.exception("newsletter: top_gainers failed") gainers = [] # 2. Pick top gainer for AI analysis ai_block_html = "" headline_coin_name = "" if gainers: top_pick = gainers[0] try: ai = await ai_analyze(AnalyzeRequest(coin_id=top_pick["id"], horizon="1m")) sig = ai["analysis"]["signal"] sig_color = "#059669" if sig == "BUY" else ("#E11D48" if sig == "SELL" else "#52525B") sig_label = {"BUY": "BELI", "SELL": "JUAL", "HOLD": "TAHAN"}.get(sig, sig) headline_coin_name = ai["coin"]["name"] ai_block_html = f"""
SINYAL AI · {ai['analysis']['confidence']}% CONFIDENCE
{sig_label}
{ai['coin']['name']} ({(ai['coin']['symbol'] or '').upper()})
Harga: {_format_usd(ai['coin'].get('current_price_usd'))} · {_format_idr(ai['coin'].get('current_price_idr'))}

{ai['analysis'].get('summary','')}

ZONA BELI
{ai['analysis'].get('buy_zone','—')}
TARGET JUAL
{ai['analysis'].get('sell_zone','—')}
STOP LOSS
{ai['analysis'].get('stop_loss','—')}

Timing: {ai['analysis'].get('timing_advice','—')}
""" except Exception: logger.exception("newsletter: ai_analyze failed") ai_block_html = "" # 3. Gainers table rows rows_html = "" for i, c in enumerate(gainers): change = c.get("price_change_percentage_24h_in_currency", 0) or 0 rows_html += f""" {i+1} {c.get('name')} {c.get('symbol')} {_format_usd(c.get('current_price'))} {_format_pct(change)} """ # 4. Exchange rate try: ex = await exchange_rate() usd_per_btc = ex.get("usd") idr_per_btc = ex.get("idr") usd_to_idr = (idr_per_btc / usd_per_btc) if usd_per_btc and idr_per_btc else None ex_html = f"1 USD = {_format_idr(usd_to_idr)}" if usd_to_idr else "Rate unavailable" except Exception: ex_html = "Rate unavailable" today_str = datetime.now(JAKARTA_TZ).strftime("%A, %d %B %Y") headline = f"☕ {headline_coin_name} hari ini · " if headline_coin_name else "" subject = f"{headline}Top Movers & AI Signal — {datetime.now(JAKARTA_TZ).strftime('%d %b')}" html = f""" CryptoEditorial Daily
CryptoEditorial
Daily Edition · {today_str}
 VOL.01
Newsletter Edisi Pagi

Pemenang
Hari Ini.

Top 5 koin dengan kenaikan tertinggi 24 jam terakhir, plus 1 sinyal AI dari Gemini 3 Pro untuk koin terdepan.

KURS · {ex_html}
Section 01 — Top 5 Gainers (24h)
{rows_html or ''}
# Coin Harga 24h
Data sedang dimuat ulang…
Section 02 — AI Pick of the Day · Gemini 3 Pro
{ai_block_html or '

AI signal tidak tersedia hari ini.

'}
Buka AI Analyzer →
Bukan nasihat finansial. Konten ini bersifat edukasi. Harga crypto sangat volatil. Selalu lakukan riset Anda sendiri (DYOR).
© CryptoEditorial · Vol.01 · 2026
Berhenti berlangganan
""" return {"subject": subject, "html": html} async def send_newsletter_to_subscribers() -> dict: """Build newsletter once and send to all active subscribers.""" if not RESEND_API_KEY: logger.warning("RESEND_API_KEY not configured — skipping newsletter") return {"sent": 0, "errors": 0, "skipped": True} subs = await db.subscribers.find({"active": True}, {"_id": 0}).to_list(2000) if not subs: return {"sent": 0, "errors": 0, "subscribers": 0} try: nl = await build_newsletter_html() except Exception as e: logger.exception("Failed to build newsletter") return {"sent": 0, "errors": 1, "error": str(e)} sent, errors = 0, 0 last_error = None for s in subs: email = s["email"] # Replace unsubscribe placeholder per recipient html_personal = nl["html"].replace("{{EMAIL}}", email) try: params = { "from": f"{SENDER_NAME} <{SENDER_EMAIL}>", "to": [email], "subject": nl["subject"], "html": html_personal, } await asyncio.to_thread(resend.Emails.send, params) sent += 1 except Exception as e: errors += 1 last_error = str(e)[:200] logger.error(f"newsletter send failed to {email}: {e}") await db.newsletter_runs.insert_one({ "run_at": datetime.now(timezone.utc).isoformat(), "sent": sent, "errors": errors, "subscribers": len(subs), "subject": nl["subject"], "last_error": last_error, }) return {"sent": sent, "errors": errors, "subscribers": len(subs), "last_error": last_error} @api_router.post("/newsletter/subscribe") async def newsletter_subscribe(payload: SubscribeRequest): email = payload.email.lower() existing = await db.subscribers.find_one({"email": email}) if existing and existing.get("active"): return {"ok": True, "already_subscribed": True} token = secrets.token_urlsafe(24) if existing: await db.subscribers.update_one( {"email": email}, {"$set": {"active": True, "unsubscribe_token": token, "resubscribed_at": datetime.now(timezone.utc).isoformat()}}, ) else: await db.subscribers.insert_one({ "id": str(uuid.uuid4()), "email": email, "active": True, "unsubscribe_token": token, "subscribed_at": datetime.now(timezone.utc).isoformat(), }) return {"ok": True, "subscribed": True} @api_router.get("/newsletter/unsubscribe", response_class=HTMLResponse) async def newsletter_unsubscribe(email: str): email = email.lower() result = await db.subscribers.update_one({"email": email}, {"$set": {"active": False, "unsubscribed_at": datetime.now(timezone.utc).isoformat()}}) msg = "Anda berhasil berhenti berlangganan." if result.matched_count else "Email tidak ditemukan." return HTMLResponse(f"""
CryptoEditorial

{msg}

Kembali ke Beranda
""") @api_router.get("/newsletter/preview", response_class=HTMLResponse) async def newsletter_preview(): """Preview today's newsletter in browser.""" nl = await build_newsletter_html() return HTMLResponse(nl["html"]) @api_router.post("/newsletter/send-now") async def newsletter_send_now(user: dict = Depends(get_current_user)): """Admin-only: trigger newsletter send to all subscribers immediately.""" if user.get("role") != "admin": raise HTTPException(status_code=403, detail="Admin only") result = await send_newsletter_to_subscribers() return result @api_router.post("/newsletter/test-send") async def newsletter_test_send(payload: SubscribeRequest, user: dict = Depends(get_current_user)): """Admin-only: send a single test newsletter to the given email.""" if user.get("role") != "admin": raise HTTPException(status_code=403, detail="Admin only") if not RESEND_API_KEY: raise HTTPException(status_code=500, detail="RESEND_API_KEY not configured") nl = await build_newsletter_html() email = payload.email.lower() html_personal = nl["html"].replace("{{EMAIL}}", email) try: params = { "from": f"{SENDER_NAME} <{SENDER_EMAIL}>", "to": [email], "subject": "[TEST] " + nl["subject"], "html": html_personal, } result = await asyncio.to_thread(resend.Emails.send, params) return {"ok": True, "id": result.get("id"), "to": email} except Exception as e: raise HTTPException(status_code=502, detail=f"Resend error: {str(e)[:300]}") # ============= ROOT ============= @api_router.get("/") async def root(): return {"message": "CryptoEditorial API", "status": "ok"} # ============= STARTUP ============= async def seed_admin(): existing = await db.users.find_one({"email": ADMIN_EMAIL.lower()}) if existing is None: await db.users.insert_one( { "id": str(uuid.uuid4()), "email": ADMIN_EMAIL.lower(), "name": "Admin", "role": "admin", "password_hash": hash_password(ADMIN_PASSWORD), "created_at": datetime.now(timezone.utc).isoformat(), } ) logger.info("Admin user seeded.") elif not verify_password(ADMIN_PASSWORD, existing["password_hash"]): await db.users.update_one( {"email": ADMIN_EMAIL.lower()}, {"$set": {"password_hash": hash_password(ADMIN_PASSWORD)}}, ) logger.info("Admin password updated.") @app.on_event("startup") async def on_startup(): try: await db.users.create_index("email", unique=True) await db.users.create_index("id", unique=True) await db.holdings.create_index("user_id") await db.login_attempts.create_index("identifier") await db.subscribers.create_index("email", unique=True) await seed_admin() # Schedule daily newsletter at NEWSLETTER_HOUR_JAKARTA in Asia/Jakarta global scheduler scheduler = AsyncIOScheduler(timezone=JAKARTA_TZ) scheduler.add_job( send_newsletter_to_subscribers, CronTrigger(hour=NEWSLETTER_HOUR_JAKARTA, minute=0, timezone=JAKARTA_TZ), id="daily_newsletter", replace_existing=True, misfire_grace_time=3600, ) scheduler.start() logger.info(f"Newsletter scheduler started — daily at {NEWSLETTER_HOUR_JAKARTA:02d}:00 Asia/Jakarta") logger.info("Startup complete.") except Exception as e: logger.exception(f"Startup error: {e}") @app.on_event("shutdown") async def on_shutdown(): try: if "scheduler" in globals() and scheduler.running: scheduler.shutdown(wait=False) except Exception: pass client.close() # ============= App wiring ============= app.include_router(api_router) # CORS — explicit origin needed when allow_credentials=True (browsers reject "*") allowed_origins = [FRONTEND_URL, "http://localhost:3000"] app.add_middleware( CORSMiddleware, allow_origins=allowed_origins, allow_credentials=True, allow_methods=["*"], allow_headers=["*"], )