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

"""
Cookie管理视图 — 插件版

URL 前缀：/system/cookie
模型和Schema从插件内部导入，避免循环依赖。
"""
import os
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

# 从插件内部导入模型和Schema（避免在applications.models中导入导致循环依赖）
from plugins.cookieManager.models import CookieManager
from plugins.cookieManager.schemas import CookieManageSchema

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


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


@bp.get('/data')
@limiter.limit("60 per minute")
@authorize("system:cookie:main")
def data():
    keyword = str_escape(request.args.get('keyword', type=str))
    website_id = request.args.get('website_id', type=int)

    query = CookieManager.query.filter()
    if keyword:
        query = query.filter(
            db.or_(
                CookieManager.name.contains(keyword),
                CookieManager.website_name.contains(keyword),
                CookieManager.website_url.contains(keyword),
            )
        )
    if website_id:
        query = query.filter(CookieManager.website_id == website_id)

    items = query.order_by(CookieManager.id.desc()).layui_paginate()
    return table_api(
        msg="请求成功",
        data=curd.model_to_dicts(schema=CookieManageSchema, data=items.items),
        count=items.total,
    )


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


@bp.post('/save')
@authorize("system:cookie:add", log=True)
def save():
    req_json = request.get_json(force=True, silent=True) or {}
    website_id = int(req_json.get('website_id') or 0)
    website_name = str_escape(req_json.get('website_name') or '')
    website_url = str_escape(req_json.get('website_url') or '')
    name = str_escape(req_json.get('name') or '')
    value = req_json.get('value') or ''
    domain = str_escape(req_json.get('domain') or '')
    path = str_escape(req_json.get('path') or '/')
    expiry_str = req_json.get('expiry') or ''
    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)

    if not all([website_id, name, value]):
        return fail_api(msg="网站ID/Cookie名称/值不得为空")

    # 解析过期时间
    expiry = None
    if expiry_str:
        try:
            expiry = datetime.datetime.strptime(expiry_str, '%Y-%m-%d %H:%M:%S')
        except ValueError:
            try:
                expiry = datetime.datetime.fromisoformat(expiry_str)
            except ValueError:
                pass

    try:
        cookie = CookieManager(
            website_id=website_id,
            website_name=website_name,
            website_url=website_url,
            name=name,
            value=value,
            domain=domain,
            path=path,
            expiry=expiry,
            secure=secure,
            http_only=http_only,
            enable=enable,
            create_by=current_user.username if current_user.is_authenticated else 'admin',
        )
        db.session.add(cookie)
        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:edit", log=True)
def edit():
    cookie_id = request.args.get('cookieId', type=int)
    cookie = curd.get_one_by_id(CookieManager, cookie_id)
    if not cookie:
        return fail_api(msg="Cookie不存在")
    return render_template('system/cookie/edit.html', cookie=cookie)


@bp.put('/update')
@authorize("system:cookie:edit", log=True)
def update():
    req_json = request.get_json(force=True, silent=True) or {}
    cookie_id = req_json.get('cookieId')
    cookie = curd.get_one_by_id(CookieManager, cookie_id)
    if not cookie:
        return fail_api(msg="Cookie不存在")

    website_id = int(req_json.get('website_id') or cookie.website_id)
    website_name = str_escape(req_json.get('website_name') or cookie.website_name)
    website_url = str_escape(req_json.get('website_url') or cookie.website_url)
    name = str_escape(req_json.get('name') or '')
    value = req_json.get('value') or ''
    domain = str_escape(req_json.get('domain') or cookie.domain)
    path = str_escape(req_json.get('path') or cookie.path)
    secure = int(req_json.get('secure') or cookie.secure)
    http_only = int(req_json.get('http_only') or cookie.http_only)
    enable = int(req_json.get('enable') or cookie.enable)

    if not all([website_id, name, value]):
        return fail_api(msg="网站ID/Cookie名称/值不得为空")

    # 解析过期时间
    expiry_str = req_json.get('expiry') or ''
    expiry = None
    if expiry_str:
        try:
            expiry = datetime.datetime.strptime(expiry_str, '%Y-%m-%d %H:%M:%S')
        except ValueError:
            try:
                expiry = datetime.datetime.fromisoformat(expiry_str)
            except ValueError:
                pass

    try:
        cookie.website_id = website_id
        cookie.website_name = website_name
        cookie.website_url = website_url
        cookie.name = name
        cookie.value = value
        cookie.domain = domain
        cookie.path = path
        cookie.expiry = expiry
        cookie.secure = secure
        cookie.http_only = http_only
        cookie.enable = enable
        db.session.commit()
    except Exception as e:
        db.session.rollback()
        return fail_api(msg=f"更新失败：{e}")
    return success_api(msg="更新成功")


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


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


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