AIbijia/
├── assets/           # Static assets (banner, images)
├── data/             # Price data / scraped results (JSON/CSV)
├── scrapers/         # Platform price scrapers
├── frontend/         # Website UI (aibijia.org)
└── SKILL.md

git clone https://github.com/ka-pi-ba-la/AIbijia.git
cd AIbijia

pip install -r requirements.txt

npm install

---

import requests
from bs4 import BeautifulSoup
import json
from datetime import datetime

class TokenPriceScraper:
    """
    Base scraper for AI token price platforms.
    Each platform subclasses this with custom parsing.
    """
    
    def __init__(self, platform_name: str, base_url: str):
        self.platform_name = platform_name
        self.base_url = base_url
        self.session = requests.Session()
        self.session.headers.update({
            "User-Agent": "Mozilla/5.0 (compatible; Aibijia/1.0)"
        })
    
    def fetch_page(self, url: str) -> BeautifulSoup:
        resp = self.session.get(url, timeout=10)
        resp.raise_for_status()
        return BeautifulSoup(resp.text, "html.parser")
    
    def parse_prices(self, soup: BeautifulSoup) -> list[dict]:
        """Override in subclass to extract price data."""
        raise NotImplementedError
    
    def scrape(self) -> list[dict]:
        soup = self.fetch_page(self.base_url)
        prices = self.parse_prices(soup)
        
        # Annotate with metadata
        for item in prices:
            item["platform"] = self.platform_name
            item["scraped_at"] = datetime.utcnow().isoformat()
        
        return prices


class KawangScraper(TokenPriceScraper):
    """Example scraper for a 卡网 (card platform)."""
    
    def parse_prices(self, soup: BeautifulSoup) -> list[dict]:
        results = []
        
        # Adapt selectors to target platform's HTML structure
        for card in soup.select(".product-card"):
            name = card.select_one(".product-name")
            price = card.select_one(".product-price")
            stock = card.select_one(".product-stock")
            
            if name and price:
                results.append({
                    "name": name.get_text(strip=True),
                    "price_cny": float(
                        price.get_text(strip=True)
                             .replace("¥", "")
                             .replace(",", "")
                    ),
                    "in_stock": stock and "有货" in stock.get_text(),
                })
        
        return results

import asyncio
import aiohttp
from dataclasses import dataclass

@dataclass
class PriceListing:
    token_type: str
    platform: str
    price_cny: float
    in_stock: bool
    url: str
    scraped_at: str

async def aggregate_all_platforms(platforms: list[TokenPriceScraper]) -> list[PriceListing]:
    """
    Run all scrapers concurrently and merge results.
    """
    results = []
    
    async def run_scraper(scraper):
        loop = asyncio.get_event_loop()
        # Run sync scraper in thread pool
        data = await loop.run_in_executor(None, scraper.scrape)
        return data
    
    tasks = [run_scraper(p) for p in platforms]
    all_data = await asyncio.gather(*tasks, return_exceptions=True)
    
    for platform_data in all_data:
        if isinstance(platform_data, Exception):
            print(f"Scraper error: {platform_data}")
            continue
        results.extend(platform_data)
    
    return results


def find_cheapest(listings: list[PriceListing], token_type: str) -> list[PriceListing]:
    """Filter and sort by price for a specific token type."""
    filtered = [
        l for l in listings
        if token_type.lower() in l.token_type.lower()
        and l.in_stock
    ]
    return sorted(filtered, key=lambda x: x.price_cny)

all_listings = await aggregate_all_platforms(platforms)
cheapest = find_cheapest(all_listings, "ChatGPT Plus")

print("Cheapest ChatGPT Plus CDKs:")
for listing in cheapest[:5]:
    print(f"  ¥{listing.price_cny} — {listing.platform}")

---

import requests
import os

AIBIJIA_API = "https://aibijia.org/api"  # hypothetical endpoint

def submit_vendor(vendor_info: dict) -> dict:
    """
    Submit a new vendor/price source for review.
    
    vendor_info keys:
      - name: str          Vendor/platform name
      - url: str           Purchase URL
      - token_type: str    e.g. "ChatGPT Plus CDK"
      - price_cny: float   Current price in RMB
      - notes: str         Optional reliability notes
    """
    resp = requests.post(
        f"{AIBIJIA_API}/submit",
        json=vendor_info,
        headers={
            "Content-Type": "application/json",
            # Use env var if auth is required:
            "Authorization": f"Bearer {os.environ.get('AIBIJIA_API_KEY', '')}",
        },
        timeout=10,
    )
    resp.raise_for_status()
    return resp.json()

---

import json
import os
from pathlib import Path
from datetime import datetime

DATA_DIR = Path("./data")

def save_price_snapshot(listings: list[dict], token_type: str):
    """Save a timestamped price snapshot to data/."""
    DATA_DIR.mkdir(exist_ok=True)
    
    date_str = datetime.utcnow().strftime("%Y-%m-%d")
    filename = DATA_DIR / f"{token_type.replace(' ', '_')}_{date_str}.json"
    
    snapshot = {
        "token_type": token_type,
        "captured_at": datetime.utcnow().isoformat(),
        "count": len(listings),
        "listings": listings,
    }
    
    with open(filename, "w", encoding="utf-8") as f:
        json.dump(snapshot, f, ensure_ascii=False, indent=2)
    
    print(f"Saved {len(listings)} listings to {filename}")


def load_latest_snapshot(token_type: str) -> dict | None:
    """Load the most recent snapshot for a token type."""
    pattern = f"{token_type.replace(' ', '_')}_*.json"
    files = sorted(DATA_DIR.glob(pattern), reverse=True)
    
    if not files:
        return None
    
    with open(files[0], encoding="utf-8") as f:
        return json.load(f)

undefined

---

undefined

# Proxy (optional, for bot detection avoidance)
"proxy": os.environ.get("HTTP_PROXY", None),

# Data output
"data_dir": os.environ.get("DATA_DIR", "./data"),

# Notifications (Telegram)
"telegram_bot_token": os.environ.get("TELEGRAM_BOT_TOKEN"),
"telegram_channel_id": os.environ.get("TELEGRAM_CHANNEL_ID"),

# Price alert threshold (alert if price drops below X CNY)
"alert_price_threshold": float(os.environ.get("ALERT_PRICE_CNY", "30.0")),

undefined

undefined

---

import os
import asyncio
from telegram import Bot

async def send_price_alert(listings: list[dict], threshold: float):
    """
    Send Telegram alert when ChatGPT Plus CDK drops below threshold price.
    """
    bot = Bot(token=os.environ["TELEGRAM_BOT_TOKEN"])
    channel = os.environ["TELEGRAM_CHANNEL_ID"]
    
    cheap = [l for l in listings if l["price_cny"] <= threshold and l["in_stock"]]
    
    if not cheap:
        return
    
    lines = [f"🔥 低价预警！ChatGPT Plus CDK ≤ ¥{threshold}\n"]
    for l in cheap[:5]:
        lines.append(f"• ¥{l['price_cny']} — {l['platform']}")
    
    await bot.send_message(
        chat_id=channel,
        text="\n".join(lines),
        disable_web_page_preview=True,
    )

asyncio.run(send_price_alert(all_listings, threshold=30.0))

undefined

---

import time
import random

def polite_get(session, url: str, min_delay=1.5, max_delay=4.0) -> str:
    """Add random delay between requests to avoid rate limiting."""
    time.sleep(random.uniform(min_delay, max_delay))
    resp = session.get(url, timeout=10)
    resp.raise_for_status()
    return resp.text

// data/sources.json entry format
{
  "id": "vendor_slug",
  "name": "平台名称",
  "url": "https://example-card-site.com",
  "token_types": ["ChatGPT Plus CDK", "Claude API"],
  "verified": false,
  "submitted_by": "github_username",
  "notes": "24h售后，微信群支持"
}