HttpOpenFile.py

import os
import json
import shutil
import platform
import http.server
import socketserver
import subprocess
from urllib.parse import urlparse, unquote_plus

def load_config(config_file='config.json'):
    """
    加载配置文件
    """
    default_config = {
        "base_dir": r"C:/Users/YourUserName/OneDrive",
        "host": "localhost",
        "port": 8000,
        "allowed_clients": ["127.0.0.1", "localhost"]
    }
    
    try:
        with open(config_file, 'r', encoding='utf-8') as f:
            config = json.load(f)
            # 用配置文件中的值更新默认配置
            default_config.update(config)
    except FileNotFoundError:
        print(f"警告: 配置文件 {config_file} 未找到，使用默认配置。")
    except json.JSONDecodeError as e:
        print(f"错误: 配置文件 {config_file} 格式错误: {e}，使用默认配置。")
    
    # 确保 base_dir 是绝对路径
    if not os.path.isabs(default_config["base_dir"]):
        default_config["base_dir"] = os.path.abspath(default_config["base_dir"])
        
    return default_config

# 加载配置
CONFIG = load_config()
BASE_DIR = CONFIG['base_dir']
HOST = CONFIG['host']
PORT = CONFIG['port']
ALLOWED_CLIENTS = CONFIG['allowed_clients']

class SimpleHTTPRequestHandler(http.server.SimpleHTTPRequestHandler):
    def do_GET(self):
        # 解析 URL
        parsed = urlparse(self.path)
        query = unquote_plus(parsed.query)

        # 检查路径是否为 '/open'
        if parsed.path != '/open':
            self.send_error(401, "Unauthorized")
            return

        # 设置 http 响应状态码
        self.send_response(200)

        # 设置 http 响应头
        self.send_header('Content-type', 'text/html;charset=utf-8')
        self.end_headers()

        # 设置 html 样式
        self.wfile.write(r"<style>body{padding:20vh}</style>".encode())
        self.wfile.write(r"<style>div,green,red,small{display:block;text-align:center}</style>".encode())
        self.wfile.write(r"<style>div{margin-bottom:20px}</style>".encode())
        self.wfile.write(r"<style>green{color:green;font-weight:bold}</style>".encode())
        self.wfile.write(r"<style>red{color:red;font-weight:bold}</style>".encode())

        # 检查请求来源
        if self.client_address[0] not in ALLOWED_CLIENTS:
            self.wfile.write("<red>非本机请求，拒绝访问 *_*</red>".encode())
            return

        # 处理查询路径
        if not query:
            self.wfile.write("<red>未指定文件路径 *_*</red>".encode())
            return

        query_path = unquote_plus(query)

        # 安全检查：防止路径遍历攻击
        requested_full_path = os.path.abspath(os.path.join(BASE_DIR, query_path))
        if not requested_full_path.startswith(os.path.abspath(BASE_DIR)):
            self.wfile.write("<red>安全错误：无权访问该路径外的文件 *_*</red>".encode())
            return

        # 检查目标路径是否存在
        if not os.path.exists(requested_full_path):
            self.wfile.write("<red>目录或文件不存在 *_*</red>".encode())
            return

        self.wfile.write(f"<div>{requested_full_path}</div>".encode())

        # 检查是目录还是文件
        if os.path.isdir(requested_full_path):
            print(">>>> is dir  >>>> " + requested_full_path)
        elif os.path.isfile(requested_full_path):
            print(">>>> is file >>>> " + requested_full_path)
        else:
            self.wfile.write("<red>目录或文件不存在 *_*</red>".encode())
            return

        # 获取系统信息
        system = platform.system()

        try:
            # 打开文件或目录，分系统执行
            if system == "Windows":
                os.startfile(requested_full_path)
            elif system == "Darwin":
                subprocess.call(['open', requested_full_path])
            elif system == "Linux":
                if shutil.which("xdg-open") is not None:
                    subprocess.call(['xdg-open', requested_full_path])
                else:
                    self.wfile.write(("<red>请先<a href='https://www.bing.com/search?q=xdg-utils%20%E5%AE%89%E8%A3%85'>" +
                                      "安装 xdg-utils</a>，然后重试 *_*</red>").encode())
                    return
            else:
                self.wfile.write(f"<red>不支持的操作系统：{system} *_*</red>".encode())
                return

        except Exception as e:
            self.wfile.write("<red>打开失败 *_*</red><br/>".encode())
            self.wfile.write(f"<small>{e}</small>".encode())
            return

        self.wfile.write("<green>已经在相关应用中打开，请关闭本窗口 ^_^</green>".encode())
        self.wfile.write("<script>window.close()</script>".encode())


# 使用配置中的主机和端口启动服务器
with socketserver.TCPServer((HOST, PORT), SimpleHTTPRequestHandler) as httpd:
    print(f"serving at {HOST}:{PORT}, base directory: {BASE_DIR}")
    httpd.serve_forever()