python自带免费ssl证书的企业网站前后台单文件代码

代码语言:python

所属分类:其他

代码描述:python自带免费ssl证书的企业网站前后台单文件代码

代码标签: python 自带 免费 ssl 证书 企业 网站 前 后台 单文件 代码

下面为部分代码预览,完整代码请点击下载或在bfwstudio webide中打开

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
企业网站管理系统 · 单文件版
用法:
    python3 site.py example.com --email admin@example.com
    python3 site.py example.com www.example.com --email admin@example.com
    python3 site.py example.com --email a@b.com --http-port 8080 --https-port 8443 --no-ssl

依赖:
    pip install flask cryptography acme josepy
"""

import os
import re
import sys
import ssl
import time
import html
import sqlite3
import secrets
import argparse
import threading
import datetime
import traceback
from functools import wraps

from flask import (Flask, request, redirect, session,
                   render_template, Response, abort)
from jinja2 import DictLoader
from werkzeug.security import generate_password_hash, check_password_hash
from werkzeug.serving import make_server

# ------------------------------------------------------------------
# 可选:ACME / 证书依赖
# ------------------------------------------------------------------
try:
    from cryptography import x509
    from cryptography.x509.oid import NameOID
    from cryptography.hazmat.primitives import hashes, serialization
    from cryptography.hazmat.primitives.asymmetric import rsa
    from cryptography.hazmat.backends import default_backend
    from acme import client as acme_client
    from acme import messages as acme_messages
    from acme.errors import ConflictError
    from josepy import jwk
    HAS_ACME = True
except Exception as _e:  # noqa
    HAS_ACME = False
    _ACME_IMPORT_ERR = str(_e)


# ------------------------------------------------------------------
# 全局配置
# ------------------------------------------------------------------
DATA_DIR = os.path.abspath('./site_data')
SSL_DIR = os.path.join(DATA_DIR, 'ssl')
DB_PATH = os.path.join(DATA_DIR, 'site.db')
CERT_PATH = os.path.join(SSL_DIR, 'fullchain.pem')
KEY_PATH = os.path.join(SSL_DIR, 'privkey.pem')

HTTP_PORT = 80
HTTPS_PORT = 443
USE_SSL = True
DOMAINS = []
EMAIL = ''
ACME_STAGING = False
RENEW_DAYS = 30          # 剩余天数低于此值就续期
RENEW_CHECK_INTERVAL = 12 * 3600   # 每 12 小时检查一次

# ACME HTTP-01 challenge token -> key authorization
acme_challenges = {}
acme_lock = threading.Lock()


# ------------------------------------------------------------------
# 通用工具
# ------------------------------------------------------------------
def log(msg):
    ts = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    print('[{}] {}'.format(ts, msg))
    sys.stdout.flush()


def ensure_dir(path):
    if path and not os.path.exists(path):
        os.makedirs(path, exist_ok=True)


def save_file(path, data, mode='wb'):
    ensure_dir(os.path.dirname(os.path.abspath(path)))
    with open(path, mode) as f:
        f.write(data)


def load_file(path, mode='rb'):
    with open(path, mode) as f:
        return f.read()


def slugify(text, fallback='item'):
    text = (text or '').strip().lower()
    text = re.sub(r'[^\w\u4e00-\u9fa5]+', '-', text)
    text = re.sub(r'-+', '-', text).strip('-')
    return text or fallback


# ------------------------------------------------------------------
# 数据库
# ------------------------------------------------------------------
def db_connect():
    conn = sqlite3.connect(DB_PATH, timeout=15)
    conn.row_factory = sqlite3.Row
    return conn


def db_query(sql, params=()):
    conn = db_connect()
    try:
        cur = conn.cursor()
        cur.execute(sql, params)
        return [dict(r) for r in cur.fetchall()]
    finally:
        conn.close()


def db_one(sql, params=()):
    rows = db_query(sql, params)
    return rows[0] if rows else None


def db_exec(sql, params=()):
    conn = db_connect()
    try:
        cur = conn.cursor()
        cur.execute(sql, params)
        conn.commit()
        return cur.lastrowid
    finally:
        conn.close()


def get_settings():
    rows = db_query('SELECT key, value FROM settings')
    return {r['key']: r['value'] for r in rows}


def get_setting(key, default=''):
    row = db_one('SELECT value FROM settings WHERE key=?', (key,))
    return row['value'] if row else default


def set_setting(key, value):
    db_exec('INSERT INTO settings (key, value) VALUES (?, ?) '
            'ON CONFLICT(key) DO UPDATE SET value=excluded.value', (key, value))


def init_db():
    ensure_dir(DATA_DIR)
    conn = db_connect()
    c = conn.cursor()
    c.executescript('''
    CREATE TABLE IF NOT EXISTS settings (
        key   TEXT PRIMARY KEY,
        value TEXT
    );
    CREATE TABLE IF NOT EXISTS categories (
        id          INTEGER PRIMARY KEY AUTOINCREMENT,
        slug        TEXT UNIQUE NOT NULL,
        name        TEXT NOT NULL,
        type        TEXT DEFAULT 'article',   -- article / product / service / page
        description TEXT DEFAULT '',
        content     TEXT DEFAULT '',          -- 仅 type=page 时使用
        icon        TEXT DEFAULT '',
        sort_order  INTEGER DEFAULT 0,
        is_show     INTEGER DEFAULT 1
    );
    CREATE TABLE IF NOT EXISTS posts (
        id           INTEGER PRIMARY KEY AUTOINCREMENT,
        category_id  INTEGER NOT NULL,
        title        TEXT NOT NULL,
        slug         TEXT NOT NULL,
        summary      TEXT DEFAULT '',
        content      TEXT DEFAULT '',
        cover        TEXT DEFAULT '',
        author       TEXT DEFAULT '',
        created_at   TEXT,
        updated_at   TEXT,
        sort_order   INTEGER DEFAULT 0,
        is_published INTEGER DEFAULT 1,
        views        INTEGER DEFAULT 0
    );
    CREATE TABLE IF NOT EXISTS users (
        id            INTEGER PRIMARY KEY AUTOINCREMENT,
        username      TEXT UNIQUE NOT NULL,
        password_hash TEXT NOT NULL,
        created_at    TEXT
    );
    CREATE INDEX IF NOT EXISTS idx_posts_category ON posts(category_id);
    CREATE INDEX IF NOT EXISTS idx_posts_slug     ON posts(slug);
    CREATE INDEX IF NOT EXISTS idx_cats_slug      ON categories(slug);
    ''')

    # 默认站点配置
    defaults = {
        'site_name':        '云启科技',
        'site_subtitle':    '专业 · 创新 · 共赢',
        'site_keywords':    '企业网站,科技公司,解决方案,产品服务',
        'site_description': '云启科技致力于为企业客户提供专业的产品与技术服务。',
        'banner_title':     '专业解决方案提供商',
        'banner_subtitle':  '以技术驱动创新,为每一家企业创造真实价值',
        'about_title':      '关于我们',
        'about_content':    '云启科技成立于 2015 年,是一家专注于企业数字化解决方案的高新技术企业。'
                            '我们为制造、零售、金融、物流等行业客户提供产品研发、系统集成与技术支持服务。',
        'contact_phone':    '400-888-8888',
        'contact_email':    'contact@example.com',
        'contact_address':  '北京市朝阳区科技园区 100 号',
        'contact_hours':    '周一至周五 9:00 - 18:00',
        'icp':              '',
        'copyright':        '© 2024 云启科技 版权所有',
        'session_key':      secrets.token_hex(32),
        'site_theme':       'blue',
    }
    for k, v in defaults.items():
        c.execute('INSERT OR IGNORE INTO settings (key, value) VALUES (?, ?)', (k, v))

    # 默认管理员
    if c.execute('SELECT COUNT(*) FROM users').fetchone()[0] == 0:
        c.execute('INSERT INTO users (username, password_hash, created_at) VALUES (?,?,?)',
                  ('admin', generate_password_hash('admin123'),
                   datetime.datetime.now().isoformat()))

    # 默认栏目
    if c.execute('SELECT COUNT(*) FROM categories').fetchone()[0] == 0:
        cats = [
            ('news',     '新闻资讯', 'article', 1, '公司新闻与行业动态', ''),
            ('products', '产品中心', 'product', 2, '我们的核心产品', ''),
            ('services', '服务支持', 'service', 3, '专业服务与技术支持', ''),
            ('about',    '关于我们', 'page',    4, '公司介绍',
             '云启科技是一家专注于企业数字化解决方案的高新技术企业。'),
            ('contact',  '联系我们', 'page',    5, '联系方式',
             '<p>电话:400-888-8888</p><p>邮箱:contact@example.com</p>'
             '<p>地址:北京市朝阳区科技园区 100 号</p>'),
        ]
        for slug, name, typ, order, desc, content in cats:
            c.execute('INSERT INTO categories (slug, name, type, description, content, sort_order) '
                      'VALUES (?,?,?,?,?,?)', (slug, name, typ, desc, content, order))

    # 示例文章
    if c.execute('SELECT COUNT(*) FROM posts').fetchone()[0] == 0:
        news_id = c.execute('SELECT id FROM categories WHERE slug=?', ('news',)).fetchone()[0]
        prod_id = c.execute('SELECT id FROM categories WHERE slug=?', ('products',)).fetchone()[0]
        srv_id  = c.execute('SELECT id FROM categories WHERE slug=?', ('services',)).fetchone()[0]
        now = dat.........完整代码请登录后点击上方下载按钮下载查看

网友评论0