pear-admin-flask/plugins/cookieManager/view/cookie_site.py

"""
Cookie管理 - 登录网站视图

URL: /system/cookie-site
"""
import os
import re
from datetime import datetime
from flask import Blueprint, render_template, request
from flask_login import current_user

from applications.common import curd
from applications.common.utils.http import table_api, fail_api, success_api
from applications.common.utils.rights import authorize
from applications.common.utils.validate import str_escape
from applications.extensions import db
from applications.extensions.init_limit import limiter
from applications.models import CookieSite
from applications.schemas import CookieSiteManageSchema

dir_path = os.path.dirname(os.path.abspath(__file__))
bp = Blueprint('cookie_site', __name__,
               url_prefix='/system/cookie-site',
               template_folder=os.path.join(dir_path, '..', 'templates'))


def _normalize_url(raw):
    """网址补齐协议头：用户常省略 http(s)://，此处自动补 https://。"""
    if not raw or not str(raw).strip():
        return raw
    u = str(raw).strip()
    if re.match(r'^[a-zA-Z][a-zA-Z0-9+.\-]*://', u):
        return u
    return 'https://' + u


def _parse_expiry(raw):
    """过期时间：空值转 None；非空字符串按常见格式解析为 datetime；解析失败返回 None。"""
    if not raw or not str(raw).strip():
        return None
    raw = str(raw).strip()
    for fmt in ('%Y-%m-%d %H:%M:%S', '%Y-%m-%d %H:%M', '%Y-%m-%d'):
        try:
            return datetime.strptime(raw, fmt)
        except ValueError:
            continue
    return None


@bp.get('/')
@authorize("system:cookie-site:main")
def main():
    return render_template('system/cookie_site/main.html')


@bp.get('/data')
@limiter.limit("60 per minute")
@authorize("system:cookie-site:main")
def data():
    name = str_escape(request.args.get('name', type=str))
    domain = str_escape(request.args.get('domain', type=str))
    enable = request.args.get('enable', type=int)
    query = CookieSite.query.filter()
    if name:
        query = query.filter(CookieSite.name.contains(name))
    if domain:
        query = query.filter(CookieSite.domain.contains(domain))
    if enable is not None:
        query = query.filter(CookieSite.enable == enable)
    items = query.order_by(CookieSite.id.desc()).layui_paginate()
    return table_api(
        msg="请求成功",
        data=curd.model_to_dicts(schema=CookieSiteManageSchema, data=items.items),
        count=items.total,
    )


@bp.get('/add')
@authorize("system:cookie-site:add", log=True)
def add():
    return render_template('system/cookie_site/add.html')


@bp.post('/save')
@authorize("system:cookie-site:add", log=True)
def save():
    req_json = request.get_json(force=True, silent=True) or {}
    name = str_escape(req_json.get('name'))
    url = str_escape(req_json.get('url'))
    cookie_value = str_escape(req_json.get('cookie_value'))
    if not all([name, url, cookie_value]):
        return fail_api(msg="名称/网址/Cookie值不得为空")
    try:
        obj = CookieSite(
            name=name,
            url=_normalize_url(url),
            cookie_value=cookie_value,
            domain=str_escape(req_json.get('domain')),
            path=str_escape(req_json.get('path')) or '/',
            expiry=_parse_expiry(req_json.get('expiry')),
            secure=int(req_json.get('secure') or 0),
            http_only=int(req_json.get('http_only') or 0),
            enable=int(req_json.get('enable') or 1),
            create_by=current_user.username if current_user.is_authenticated else 'admin',
        )
        db.session.add(obj)
        db.session.commit()
    except Exception as e:
        db.session.rollback()
        return fail_api(msg=f"新增失败：{e}")
    return success_api(msg="新增成功")


@bp.get('/edit')
@authorize("system:cookie-site:edit", log=True)
def edit():
    sid = request.args.get('cookieSiteId', type=int)
    obj = curd.get_one_by_id(CookieSite, sid)
    if not obj:
        return fail_api(msg="记录不存在")
    return render_template('system/cookie_site/edit.html', obj=obj)


@bp.put('/update')
@authorize("system:cookie-site:edit", log=True)
def update():
    req_json = request.get_json(force=True, silent=True) or {}
    sid = req_json.get('cookieSiteId')
    obj = curd.get_one_by_id(CookieSite, sid)
    if not obj:
        return fail_api(msg="记录不存在")
    name = str_escape(req_json.get('name'))
    url = str_escape(req_json.get('url'))
    cookie_value = str_escape(req_json.get('cookie_value'))
    if not all([name, url, cookie_value]):
        return fail_api(msg="名称/网址/Cookie值不得为空")
    try:
        obj.name = name
        obj.url = _normalize_url(url)
        obj.cookie_value = cookie_value
        obj.domain = str_escape(req_json.get('domain'))
        obj.path = str_escape(req_json.get('path')) or '/'
        obj.expiry = _parse_expiry(req_json.get('expiry'))
        obj.secure = int(req_json.get('secure') or 0)
        obj.http_only = int(req_json.get('http_only') or 0)
        obj.enable = int(req_json.get('enable') or 1)
        db.session.commit()
    except Exception as e:
        db.session.rollback()
        return fail_api(msg=f"更新失败：{e}")
    return success_api(msg="更新成功")


@bp.delete('/remove/<int:sid>')
@authorize("system:cookie-site:remove", log=True)
def remove(sid):
    res = curd.delete_one_by_id(CookieSite, sid)
    if not res:
        return fail_api(msg="删除失败")
    return success_api(msg="删除成功")


@bp.put('/enable')
@authorize("system:cookie-site:edit", log=True)
def enable():
    req_json = request.get_json(force=True, silent=True) or {}
    sid = req_json.get('cookieSiteId')
    if not sid:
        return fail_api(msg="数据错误")
    res = curd.enable_status(CookieSite, sid)
    if not res:
        return fail_api(msg="操作失败")
    return success_api(msg="已启用")


@bp.put('/disable')
@authorize("system:cookie-site:edit", log=True)
def disable():
    req_json = request.get_json(force=True, silent=True) or {}
    sid = req_json.get('cookieSiteId')
    if not sid:
        return fail_api(msg="数据错误")
    res = curd.disable_status(CookieSite, sid)
    if not res:
        return fail_api(msg="操作失败")
    return success_api(msg="已禁用")