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

"""
Cookie管理 - 登录用户视图

URL: /system/cookie-user
"""
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 CookieUser
from applications.schemas import CookieUserManageSchema

dir_path = os.path.dirname(os.path.abspath(__file__))
bp = Blueprint('cookie_user', __name__,
               url_prefix='/system/cookie-user',
               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-user:main")
def main():
    return render_template('system/cookie_user/main.html')


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


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


@bp.post('/save')
@authorize("system:cookie-user:add", log=True)
def save():
    req_json = request.get_json(force=True, silent=True) or {}
    username = str_escape(req_json.get('username'))
    website_url = str_escape(req_json.get('website_url'))
    if not all([username, website_url]):
        return fail_api(msg="用户名/网站URL不得为空")
    try:
        obj = CookieUser(
            username=username,
            nickname=str_escape(req_json.get('nickname')),
            website_url=_normalize_url(website_url),
            website_name=str_escape(req_json.get('website_name')),
            expiry=_parse_expiry(req_json.get('expiry')),
            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-user:edit", log=True)
def edit():
    cid = request.args.get('cookieUserId', type=int)
    obj = curd.get_one_by_id(CookieUser, cid)
    if not obj:
        return fail_api(msg="记录不存在")
    return render_template('system/cookie_user/edit.html', obj=obj)


@bp.put('/update')
@authorize("system:cookie-user:edit", log=True)
def update():
    req_json = request.get_json(force=True, silent=True) or {}
    cid = req_json.get('cookieUserId')
    obj = curd.get_one_by_id(CookieUser, cid)
    if not obj:
        return fail_api(msg="记录不存在")
    username = str_escape(req_json.get('username'))
    website_url = str_escape(req_json.get('website_url'))
    if not all([username, website_url]):
        return fail_api(msg="用户名/网站URL不得为空")
    try:
        obj.username = username
        obj.nickname = str_escape(req_json.get('nickname'))
        obj.website_url = _normalize_url(website_url)
        obj.website_name = str_escape(req_json.get('website_name'))
        obj.expiry = _parse_expiry(req_json.get('expiry'))
        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:cid>')
@authorize("system:cookie-user:remove", log=True)
def remove(cid):
    res = curd.delete_one_by_id(CookieUser, cid)
    if not res:
        return fail_api(msg="删除失败")
    return success_api(msg="删除成功")


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


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