vps_web/app.py

# -*- coding: utf-8 -*-
"""云服务器价格对比 - Flask 应用"""
import io
from urllib.parse import urlencode
from flask import Flask, render_template, jsonify, request, redirect, url_for, session, send_file
from werkzeug.middleware.proxy_fix import ProxyFix
from sqlalchemy import text
from config import Config
from extensions import db
from openpyxl import Workbook
from openpyxl import load_workbook

app = Flask(__name__)
app.config.from_object(Config)
# 部署在 Nginx 等反向代理后时，信任 X-Forwarded-* 头（HTTPS、真实 IP）
app.wsgi_app = ProxyFix(app.wsgi_app, x_for=1, x_proto=1, x_host=1)
db.init_app(app)

from models import VPSPlan, Provider  # noqa: E402


def _ensure_mysql_columns():
    """为已有 MySQL 表添加缺失列，避免 1054 Unknown column。"""
    try:
        engine = db.engine
        if engine.dialect.name != "mysql":
            return
        with engine.connect() as conn:
            for col, spec in [
                ("traffic", "VARCHAR(64) NULL"),
                ("countries", "VARCHAR(255) NULL"),
                ("provider_id", "INT NULL"),
            ]:
                try:
                    conn.execute(text("ALTER TABLE vps_plans ADD COLUMN {} {}".format(col, spec)))
                    conn.commit()
                except Exception:
                    conn.rollback()
            for col, spec in [
                ("name", "VARCHAR(128) NULL"),
                ("region", "VARCHAR(128) NULL"),
                ("price_cny", "DOUBLE NULL"),
                ("price_usd", "DOUBLE NULL"),
            ]:
                try:
                    conn.execute(text("ALTER TABLE vps_plans MODIFY COLUMN {} {}".format(col, spec)))
                    conn.commit()
                except Exception:
                    conn.rollback()
    except Exception:
        pass  # 表不存在或非 MySQL 时忽略


# 启动时自动创建表（若不存在），并为已有表补列
with app.app_context():
    db.create_all()
    _ensure_mysql_columns()

ADMIN_PASSWORD = app.config["ADMIN_PASSWORD"]
SITE_URL = app.config["SITE_URL"]
SITE_NAME = app.config["SITE_NAME"]

# 国家/区域标签，供后台表单选择
COUNTRY_TAGS = [
    "中国大陆", "中国香港", "中国台湾", "美国", "日本", "新加坡", "韩国",
    "德国", "英国", "法国", "荷兰", "加拿大", "澳大利亚", "印度", "其他",
]


def admin_required(f):
    from functools import wraps
    @wraps(f)
    def wrapped(*args, **kwargs):
        if not session.get("admin_logged_in"):
            return redirect(url_for("admin_login"))
        return f(*args, **kwargs)
    return wrapped


@app.route("/")
def index():
    plans = VPSPlan.query.order_by(VPSPlan.provider, VPSPlan.price_cny).all()
    return render_template(
        "index.html",
        site_url=SITE_URL,
        site_name=SITE_NAME,
        plans_json_ld=[p.to_dict() for p in plans],
    )


@app.route("/api/plans")
def api_plans():
    plans = VPSPlan.query.order_by(VPSPlan.provider, VPSPlan.price_cny).all()
    return jsonify([p.to_dict() for p in plans])


# ---------- SEO ----------
@app.route("/sitemap.xml")
def sitemap():
    from flask import make_response
    url = SITE_URL.rstrip("/")
    xml = f'''<?xml version="1.0" encoding="UTF-8"?>
<urlset xmlns="http://www.sitemaps.org/schemas/sitemap/0.9">
  <url>
    <loc>{url}/</loc>
    <changefreq>weekly</changefreq>
    <priority>1.0</priority>
  </url>
</urlset>'''
    resp = make_response(xml)
    resp.mimetype = "application/xml"
    return resp


@app.route("/robots.txt")
def robots():
    from flask import make_response
    url = SITE_URL.rstrip("/")
    txt = f"""User-agent: *
Allow: /

Sitemap: {url}/sitemap.xml
"""
    resp = make_response(txt)
    resp.mimetype = "text/plain"
    return resp


# ---------- 后台 ----------
@app.route("/admin/login", methods=["GET", "POST"])
def admin_login():
    if request.method == "POST":
        password = request.form.get("password", "")
        if password == ADMIN_PASSWORD:
            session["admin_logged_in"] = True
            return redirect(url_for("admin_dashboard"))
        return render_template("admin/login.html", error="密码错误")
    return render_template("admin/login.html")


@app.route("/admin/logout")
def admin_logout():
    session.pop("admin_logged_in", None)
    return redirect(url_for("index"))


@app.route("/admin/api/plan/<int:plan_id>")
@admin_required
def admin_api_plan(plan_id):
    plan = VPSPlan.query.get_or_404(plan_id)
    return jsonify({
        "id": plan.id,
        "provider_id": plan.provider_id,
        "countries": plan.countries or "",
        "vcpu": plan.vcpu,
        "memory_gb": plan.memory_gb,
        "storage_gb": plan.storage_gb,
        "bandwidth_mbps": plan.bandwidth_mbps,
        "traffic": plan.traffic or "",
        "price_cny": float(plan.price_cny) if plan.price_cny is not None else None,
        "price_usd": float(plan.price_usd) if plan.price_usd is not None else None,
        "currency": plan.currency or "CNY",
        "official_url": plan.official_url or "",
    })


@app.route("/admin")
@admin_required
def admin_dashboard():
    providers = Provider.query.order_by(Provider.name).all()
    plans = VPSPlan.query.order_by(VPSPlan.provider, VPSPlan.price_cny).all()
    return render_template(
        "admin/dashboard.html",
        providers=providers,
        plans=plans,
        country_tags=COUNTRY_TAGS,
    )


# ---------- 厂商管理 ----------
@app.route("/admin/providers")
@admin_required
def admin_providers():
    providers = Provider.query.order_by(Provider.name).all()
    return render_template("admin/providers.html", providers=providers)


@app.route("/admin/provider/new", methods=["GET", "POST"])
@admin_required
def admin_provider_new():
    if request.method == "POST":
        name = request.form.get("name", "").strip()
        official_url = request.form.get("official_url", "").strip() or None
        if not name:
            return render_template("admin/provider_form.html", provider=None, error="请填写厂商名称")
        if Provider.query.filter_by(name=name).first():
            return render_template("admin/provider_form.html", provider=None, error="该厂商名称已存在")
        p = Provider(name=name, official_url=official_url)
        db.session.add(p)
        db.session.commit()
        return redirect(url_for("admin_provider_detail", provider_id=p.id))
    return render_template("admin/provider_form.html", provider=None)


@app.route("/admin/provider/<int:provider_id>")
@admin_required
def admin_provider_detail(provider_id):
    provider = Provider.query.get_or_404(provider_id)
    plans = VPSPlan.query.filter(
        (VPSPlan.provider_id == provider_id) | (VPSPlan.provider == provider.name)
    ).order_by(VPSPlan.price_cny.asc(), VPSPlan.name).all()
    providers = Provider.query.order_by(Provider.name).all()
    return render_template(
        "admin/provider_detail.html",
        provider=provider,
        plans=plans,
        providers=providers,
        country_tags=COUNTRY_TAGS,
    )


@app.route("/admin/provider/<int:provider_id>/edit", methods=["GET", "POST"])
@admin_required
def admin_provider_edit(provider_id):
    provider = Provider.query.get_or_404(provider_id)
    if request.method == "POST":
        provider.name = request.form.get("name", "").strip()
        provider.official_url = request.form.get("official_url", "").strip() or None
        if not provider.name:
            return render_template("admin/provider_form.html", provider=provider, error="请填写厂商名称")
        db.session.commit()
        return redirect(url_for("admin_provider_detail", provider_id=provider.id))
    return render_template("admin/provider_form.html", provider=provider)


@app.route("/admin/provider/<int:provider_id>/delete", methods=["POST"])
@admin_required
def admin_provider_delete(provider_id):
    provider = Provider.query.get_or_404(provider_id)
    # 将该厂商下的配置改为无厂商关联（保留配置，仅清空 provider_id）
    VPSPlan.query.filter_by(provider_id=provider_id).update({"provider_id": None})
    db.session.delete(provider)
    db.session.commit()
    return redirect(url_for("admin_providers"))


@app.route("/admin/plan/new", methods=["GET", "POST"])
@admin_required
def admin_plan_new():
    provider_id = request.args.get("provider_id", type=int)
    if request.method == "POST":
        return _save_plan(None)
    providers = Provider.query.order_by(Provider.name).all()
    return render_template(
        "admin/plan_form.html",
        plan=None,
        country_tags=COUNTRY_TAGS,
        providers=providers,
        preselected_provider_id=provider_id,
    )


@app.route("/admin/plan/<int:plan_id>/edit", methods=["GET", "POST"])
@admin_required
def admin_plan_edit(plan_id):
    plan = VPSPlan.query.get_or_404(plan_id)
    if request.method == "POST":
        return _save_plan(plan)
    return redirect(url_for("admin_dashboard"))


def _parse_optional_int(s):
    s = (s or "").strip()
    if not s:
        return None
    try:
        return int(s)
    except ValueError:
        return None


def _parse_optional_float(s):
    s = (s or "").strip()
    if not s:
        return None
    try:
        return float(s)
    except ValueError:
        return None


def _save_plan(plan):
    provider_id = request.form.get("provider_id", type=int)
    countries = request.form.get("countries", "").strip() or None
    vcpu = _parse_optional_int(request.form.get("vcpu"))
    memory_gb = _parse_optional_int(request.form.get("memory_gb"))
    storage_gb = _parse_optional_int(request.form.get("storage_gb"))
    bandwidth_mbps = _parse_optional_int(request.form.get("bandwidth_mbps"))
    traffic = request.form.get("traffic", "").strip() or None
    price_cny = _parse_optional_float(request.form.get("price_cny"))
    price_usd = _parse_optional_float(request.form.get("price_usd"))
    currency = request.form.get("currency", "CNY").strip() or "CNY"
    official_url = request.form.get("official_url", "").strip() or None

    provider = None
    if provider_id:
        provider = Provider.query.get(provider_id)
    if not provider:
        providers = Provider.query.order_by(Provider.name).all()
        return render_template(
            "admin/plan_form.html",
            plan=plan,
            country_tags=COUNTRY_TAGS,
            providers=providers,
            preselected_provider_id=provider_id,
            error="请选择厂商",
        )

    if plan is None:
        plan = VPSPlan(
            provider_id=provider.id,
            provider=provider.name,
            region=None,
            name=None,
            vcpu=vcpu,
            memory_gb=memory_gb,
            storage_gb=storage_gb,
            bandwidth_mbps=bandwidth_mbps,
            traffic=traffic,
            price_cny=price_cny,
            price_usd=price_usd,
            currency=currency,
            official_url=official_url,
            countries=countries,
        )
        db.session.add(plan)
    else:
        plan.provider_id = provider.id
        plan.provider = provider.name
        plan.region = None
        plan.name = None
        plan.vcpu = vcpu
        plan.memory_gb = memory_gb
        plan.storage_gb = storage_gb
        plan.bandwidth_mbps = bandwidth_mbps
        plan.traffic = traffic
        plan.price_cny = price_cny
        plan.price_usd = price_usd
        plan.currency = currency
        plan.official_url = official_url
        plan.countries = countries

    db.session.commit()
    # 若从厂商详情页进入添加，保存后返回该厂商详情
    from_provider_id = request.form.get("from_provider_id", type=int)
    if from_provider_id:
        return redirect(url_for("admin_provider_detail", provider_id=from_provider_id))
    return redirect(url_for("admin_dashboard"))


@app.route("/admin/plan/<int:plan_id>/delete", methods=["POST"])
@admin_required
def admin_plan_delete(plan_id):
    plan = VPSPlan.query.get_or_404(plan_id)
    db.session.delete(plan)
    db.session.commit()
    return redirect(url_for("admin_dashboard"))


# ---------- Excel 导出 / 导入 ----------
EXCEL_HEADERS = [
    "厂商", "厂商官网", "国家", "vCPU", "内存GB", "存储GB", "带宽Mbps", "流量",
    "月付人民币", "月付美元", "货币", "配置官网",
]


@app.route("/admin/export/excel")
@admin_required
def admin_export_excel():
    wb = Workbook()
    ws = wb.active
    ws.title = "配置"
    ws.append(EXCEL_HEADERS)
    plans = VPSPlan.query.order_by(VPSPlan.provider, VPSPlan.price_cny).all()
    for p in plans:
        provider_url = (p.provider_rel.official_url if p.provider_rel else "") or ""
        ws.append([
            p.provider_name,
            provider_url or "",
            p.countries or "",
            p.vcpu if p.vcpu is not None else "",
            p.memory_gb if p.memory_gb is not None else "",
            p.storage_gb if p.storage_gb is not None else "",
            p.bandwidth_mbps if p.bandwidth_mbps is not None else "",
            p.traffic or "",
            p.price_cny if p.price_cny is not None else "",
            p.price_usd if p.price_usd is not None else "",
            p.currency or "CNY",
            p.official_url or "",
        ])
    buf = io.BytesIO()
    wb.save(buf)
    buf.seek(0)
    return send_file(
        buf,
        mimetype="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
        as_attachment=True,
        download_name="vps_配置_导出.xlsx",
    )


def _plan_match(plan, row):
    """判断 DB 中的 plan 与导入行是否同一配置（厂商+配置+价格）。"""
    def eq(a, b):
        if a is None and b in (None, ""):
            return True
        if a is None or b in (None, ""):
            return a == b
        return a == b

    return (
        plan.provider_name == (row.get("厂商") or "").strip()
        and eq(plan.vcpu, _num(row.get("vCPU")))
        and eq(plan.memory_gb, _num(row.get("内存GB")))
        and eq(plan.storage_gb, _num(row.get("存储GB")))
        and eq(plan.bandwidth_mbps, _num(row.get("带宽Mbps")))
        and (plan.countries or "").strip() == (row.get("国家") or "").strip()
        and eq(plan.price_cny, _float(row.get("月付人民币")))
        and eq(plan.price_usd, _float(row.get("月付美元")))
    )


def _num(v):
    if v is None or v == "":
        return None
    try:
        return int(float(v))
    except (ValueError, TypeError):
        return None


def _float(v):
    if v is None or v == "":
        return None
    try:
        return float(v)
    except (ValueError, TypeError):
        return None


@app.route("/admin/import", methods=["GET", "POST"])
@admin_required
def admin_import():
    if request.method == "GET":
        return render_template("admin/import.html")
    f = request.files.get("file")
    if not f or not f.filename:
        return render_template("admin/import.html", error="请选择 Excel 文件")
    if not f.filename.lower().endswith(".xlsx"):
        return render_template("admin/import.html", error="请上传 .xlsx 文件")
    try:
        wb = load_workbook(io.BytesIO(f.read()), read_only=True, data_only=True)
        ws = wb.active
        rows = list(ws.iter_rows(min_row=2, values_only=True))
    except Exception as e:
        return render_template("admin/import.html", error="解析失败: {}".format(str(e)))
    headers = EXCEL_HEADERS
    parsed = []
    for row in rows:
        if not any(cell is not None and str(cell).strip() for cell in row):
            continue
        d = {}
        for i, h in enumerate(headers):
            if i < len(row):
                v = row[i]
                if v is not None and hasattr(v, "strip"):
                    v = v.strip()
                d[h] = v
            else:
                d[h] = None
        parsed.append(d)
    if not parsed:
        return render_template("admin/import.html", error="文件中没有有效数据行")
    plans = VPSPlan.query.all()
    to_add = []
    for row in parsed:
        provider_name = (row.get("厂商") or "").strip()
        if not provider_name:
            continue
        found = False
        for p in plans:
            if _plan_match(p, row):
                found = True
                break
        if not found:
            to_add.append(row)
    session["import_preview"] = to_add
    return redirect(url_for("admin_import_preview"))


@app.route("/admin/import/preview", methods=["GET", "POST"])
@admin_required
def admin_import_preview():
    to_add = session.get("import_preview") or []
    if request.method == "GET":
        return render_template("admin/import_preview.html", rows=list(enumerate(to_add)))
    selected = request.form.getlist("row_index")
    if not selected:
        to_add = session.get("import_preview") or []
        return render_template("admin/import_preview.html", rows=list(enumerate(to_add)), error="请至少勾选一行")
    to_add = session.get("import_preview") or []
    indices = set(int(x) for x in selected if x.isdigit())
    for i in indices:
        if i < 0 or i >= len(to_add):
            continue
        row = to_add[i]
        provider_name = (row.get("厂商") or "").strip()
        if not provider_name:
            continue
        provider = Provider.query.filter_by(name=provider_name).first()
        if not provider:
            provider = Provider(name=provider_name, official_url=(row.get("厂商官网") or "").strip() or None)
            db.session.add(provider)
            db.session.flush()
        plan = VPSPlan(
            provider_id=provider.id,
            provider=provider.name,
            region=None,
            name=None,
            vcpu=_num(row.get("vCPU")),
            memory_gb=_num(row.get("内存GB")),
            storage_gb=_num(row.get("存储GB")),
            bandwidth_mbps=_num(row.get("带宽Mbps")),
            traffic=(row.get("流量") or "").strip() or None,
            price_cny=_float(row.get("月付人民币")),
            price_usd=_float(row.get("月付美元")),
            currency=(row.get("货币") or "CNY").strip() or "CNY",
            official_url=(row.get("配置官网") or "").strip() or None,
            countries=(row.get("国家") or "").strip() or None,
        )
        db.session.add(plan)
    db.session.commit()
    session.pop("import_preview", None)
    return redirect(url_for("admin_dashboard") + "?" + urlencode({"msg": "已导入 {} 条配置".format(len(indices))}))


if __name__ == "__main__":
    app.run(debug=True, port=5001)