QQ空间说说和日志批量导出工具 v3

QQ空间说说和日志批量导出工具 v3

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
QQ空间说说和日志批量导出工具 v3
================================
功能:导出QQ空间的「说说」和「日志」,保存为txt文本,包含日期信息
依赖:仅Python标准库,无需pip install任何东西

使用方法:
1. 浏览器登录QQ空间
2. F12 → Network → 复制完整Cookie字符串
3. 运行本脚本,粘贴Cookie

获取完整Cookie步骤:
  ① 浏览器打开 https://user.qzone.qq.com/你的QQ号 并登录
  ② 按 F12 打开开发者工具
  ③ 切换到 Network(网络) 标签页
  ④ 刷新页面(F5)
  ⑤ 在请求列表中找到任意一个发往 qzone.qq.com 的请求,点击它
  ⑥ 在右侧 Headers(标头) 中找到 Request Headers 下的 Cookie
  ⑦ 右键复制完整的 Cookie 值(一整行很长的字符串)
  ⑧ 粘贴到脚本中运行

  或者用 Application 标签页:
  ① F12 → Application(应用) → Cookies → https://user.qzone.qq.com
  ② 把所有Cookie拼成 "key1=value1; key2=value2; ..." 格式
  ③ 至少要包含 p_skey 和 uin
"""

import json
import os
import re
import sys
import time
import urllib.request
import urllib.parse
import urllib.error
from datetime import datetime

# ============================================================
#  配置区
# ============================================================
QQ_NUMBER = "100984450"
COOKIE = ""  # 粘贴完整Cookie字符串,留空则运行时提示输入
OUTPUT_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "qqzone_export")

# 请求间隔(秒)
REQUEST_DELAY = 1.0


# ============================================================
#  Cookie 处理
# ============================================================

def extract_cookie_value(cookie_str, key):
    """从完整Cookie字符串中提取指定key的值"""
    # 匹配 key=value 的模式
    m = re.search(r'(?:^|;\s*)' + re.escape(key) + r'=([^;]+)', cookie_str)
    if m:
        return m.group(1).strip()
    return ""


def get_g_tk(p_skey):
    """根据 p_skey 计算 g_tk 签名"""
    h = 5381
    for ch in p_skey:
        h += (h << 5) + ord(ch)
    return h & 0x7FFFFFFF


def build_headers(qq_number, cookie_str):
    """构建请求头,使用完整Cookie"""
    return {
        "User-Agent": (
            "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
            "AppleWebKit/537.36 (KHTML, like Gecko) "
            "Chrome/125.0.0.0 Safari/537.36"
        ),
        "Referer": f"https://user.qzone.qq.com/{qq_number}",
        "Cookie": cookie_str,
        "Accept": "*/*",
        "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
        "Accept-Encoding": "identity",  # 不压缩,避免解压问题
        "Connection": "keep-alive",
    }


# ============================================================
#  HTTP 请求
# ============================================================

def http_get(url, headers, timeout=20):
    """用标准库发GET请求"""
    req = urllib.request.Request(url, headers=headers)
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            return resp.read().decode("utf-8", errors="replace")
    except urllib.error.HTTPError as e:
        print(f"    [HTTP {e.code}] {e.reason}")
        try:
            body = e.read().decode("utf-8", errors="replace")
            # WAF拦截会返回HTML页面
            if "waf.tencent.com" in body:
                print(f"    [WAF拦截] 腾讯WAF拒绝了请求,请检查Cookie是否完整且未过期")
            else:
                print(f"    响应内容: {body[:300]}")
        except Exception:
            pass
        return None
    except Exception as e:
        print(f"    [请求异常] {e}")
        return None


def parse_jsonp(text):
    """从JSONP响应中提取JSON数据"""
    if not text:
        return None
    text = text.strip()
    # 尝试匹配 _preloadCallback({...}) 等格式
    m = re.search(r'_\w+Callback\((.*)\)\s*;?\s*$', text, re.DOTALL)
    if m:
        try:
            return json.loads(m.group(1))
        except json.JSONDecodeError:
            pass
    # 去掉末尾分号
    text = text.rstrip(";").strip()
    if text.startswith("(") and text.endswith(")"):
        text = text[1:-1]
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        pass
    # 提取第一个 { 到最后一个 }
    first = text.find("{")
    last = text.rfind("}")
    if first != -1 and last != -1 and last > first:
        try:
            return json.loads(text[first:last + 1])
        except json.JSONDecodeError:
            pass
    return None


# ============================================================
#  文本处理
# ============================================================

def clean_html(text):
    """清理HTML标签和实体"""
    if not text:
        return ""
    text = re.sub(r'<br\s*/?>', '\n', text, flags=re.IGNORECASE)
    text = re.sub(r'<p[^>]*>', '\n', text, flags=re.IGNORECASE)
    text = re.sub(r'</p>', '\n', text, flags=re.IGNORECASE)
    text = re.sub(r'<div[^>]*>', '\n', text, flags=re.IGNORECASE)
    text = re.sub(r'</div>', '\n', text, flags=re.IGNORECASE)
    text = re.sub(r'<[^>]+>', '', text)
    entities = {
        '&nbsp;': ' ', '&lt;': '<', '&gt;': '>',
        '&amp;': '&', '&quot;': '"', '&#39;': "'",
        '&ldquo;': '"', '&rdquo;': '"',
        '&hellip;': '...', '&mdash;': '—',
    }
    for entity, char in entities.items():
        text = text.replace(entity, char)
    text = re.sub(r'&#(\d+);', lambda m: chr(int(m.group(1))), text)
    text = re.sub(r'\n{3,}', '\n\n', text)
    return text.strip()


def save_file(filepath, content):
    """保存文本到文件"""
    d = os.path.dirname(filepath)
    if d:
        os.makedirs(d, exist_ok=True)
    with open(filepath, "w", encoding="utf-8") as f:
        f.write(content)


def safe_filename(name, max_len=60):
    """生成安全的文件名"""
    name = re.sub(r'[<>:"/\\|?*\n\r\t]', '_', name)
    name = name.strip().strip('.')
    if len(name) > max_len:
        name = name[:max_len]
    return name if name else "untitled"


# ============================================================
#  说说导出
# ============================================================

# 说说API多前缀,h5优先(user.qzone被WAF拦501)
SHUOSHUO_API_PREFIXES = [
    "https://h5.qzone.qq.com/proxy/domain/taotao.qq.com",
    "https://user.qzone.qq.com/proxy/domain/taotao.qq.com",
]


def fetch_shuoshuo_page(qq_number, cookie_str, g_tk, pos, num=20):
    """获取一页说说数据,依次尝试多个API前缀"""
    params = {
        "uin": qq_number,
        "ftype": "0",
        "sort": "0",
        "pos": str(pos),
        "num": str(num),
        "replynum": "100",
        "g_tk": str(g_tk),
        "callback": "_preloadCallback",
        "code_version": "1",
        "format": "jsonp",
        "need_private_comment": "1",
    }
    headers = build_headers(qq_number, cookie_str)
    for prefix in SHUOSHUO_API_PREFIXES:
        url = prefix + "/cgi-bin/emotion_cgi_msglist_v6?" + urllib.parse.urlencode(params)
        text = http_get(url, headers)
        if not text:
            continue
        if "waf.tencent.com" in text:
            print(f"    [WAF拦截] {prefix},尝试下一个...")
            continue
        data = parse_jsonp(text)
        if data is not None:
            return data
    return None


def export_shuoshuo(qq_number, cookie_str, g_tk, output_dir):
    """导出全部说说"""
    all_items = []
    pos = 0
    num = 20

    print()
    print("=" * 60)
    print("  开始导出说说")
    print("=" * 60)

    while True:
        print(f"  获取第 {pos + 1} ~ {pos + num} 条...")
        data = fetch_shuoshuo_page(qq_number, cookie_str, g_tk, pos, num)

        if not data:
            print("  返回数据为空,停止")
            break

        code = data.get("code", -1)
        if code != 0:
            msg = data.get("message") or data.get("msg") or ""
            print(f"  接口返回 code={code}, message={msg}")
            print(f"  完整返回(前300字): {str(data)[:300]}")
            if code in [-3000, -3001, -4001]:
                print("  [提示] 登录态可能已失效,请重新获取Cookie!")
            break

        msglist = data.get("msglist") or []
        if not msglist:
            print("  没有更多说说了")
            break

        for item in msglist:
            content = clean_html(item.get("content", ""))
            ts = item.get("created_time", 0)
            if ts:
                date_str = datetime.fromtimestamp(ts).strftime("%Y-%m-%d %H:%M:%S")
            else:
                date_str = "未知日期"

            pic_urls = []
            for pic in (item.get("pic") or []):
                u = pic.get("url1") or pic.get("url") or pic.get("url2") or pic.get("url3") or ""
                if u:
                    pic_urls.append(u)

            all_items.append({
                "date": date_str,
                "content": content,
                "pics": pic_urls,
                "forward_from": item.get("rtname", ""),
            })

            preview = content[:40].replace("\n", " ") if content else "(空)"
            print(f"    [{date_str}] {preview}...")

        pos += num
        total = data.get("total", 0)
        if total and pos >= total:
            print(f"  已获取全部 {total} 条说说")
            break
        if len(msglist) < num:
            print("  已获取全部说说")
            break

        time.sleep(REQUEST_DELAY)

    # ---- 写文件 ----
    txt_path = os.path.join(output_dir, "说说.txt")
    lines = []
    lines.append("QQ空间说说导出")
    lines.append(f"QQ号: {qq_number}")
    lines.append(f"导出时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    lines.append(f"共计 {len(all_items)} 条说说")
    lines.append("=" * 60)
    lines.append("")

    for i, item in enumerate(all_items, 1):
        lines.append(f"【第{i}条】")
        lines.append(f"日期: {item['date']}")
        lines.append(f"内容: {item['content']}")
        if item["pics"]:
            lines.append("图片:")
            for j, url in enumerate(item["pics"], 1):
                lines.append(f"  图{j}: {url}")
        if item["forward_from"]:
            lines.append(f"转发自: {item['forward_from']}")
        lines.append("-" * 60)
        lines.append("")

    save_file(txt_path, "\n".join(lines))
    print(f"\n  说说已保存: {txt_path}")
    print(f"  共计 {len(all_items)} 条")
    return len(all_items)


# ============================================================
#  日志导出
# ============================================================
# 日志API域名: b.qzone.qq.com
# 列表接口: /cgi-bin/blognew/get_abs
# 详情接口: /cgi-bin/blognew/blog_output_data
# 代理前缀: user.qzone.qq.com/proxy/domain/b.qzone.qq.com

# 多个可能的代理前缀,依次尝试
BLOG_API_PREFIXES = [
    "https://user.qzone.qq.com/proxy/domain/b.qzone.qq.com",
    "https://h5.qzone.qq.com/proxy/domain/b.qzone.qq.com",
]


def try_api_endpoints(path, params, qq_number, cookie_str):
    """依次尝试多个API前缀,返回第一个成功的响应"""
    headers = build_headers(qq_number, cookie_str)
    for prefix in BLOG_API_PREFIXES:
        url = prefix + path + "?" + urllib.parse.urlencode(params)
        text = http_get(url, headers)
        if not text:
            continue
        # 检查是否是WAF页面或错误页面
        if "waf.tencent.com" in text:
            print(f"    [WAF拦截] {prefix}")
            continue
        data = parse_jsonp(text)
        if data is not None:
            return data
        # 如果JSONP解析失败但文本不为空,也返回文本(可能是纯HTML)
        if len(text) > 100:
            return text
    return None


def fetch_blog_count(qq_number, cookie_str, g_tk):
    """获取日志总数和分类信息"""
    params = {
        "inCharset": "utf-8",
        "outCharset": "utf-8",
        "format": "jsonp",
        "hostUin": qq_number,
        "uin": qq_number,
        "g_tk": str(g_tk),
        "blogType": "0",
        "reqInfo": "2",
    }
    result = try_api_endpoints("/cgi-bin/blognew/get_abs", params, qq_number, cookie_str)
    if not result or not isinstance(result, dict):
        print(f"  [调试] 获取日志分类失败")
        return 0, []
    if result.get("code") != 0:
        print(f"  [调试] 日志分类接口 code={result.get('code')}, msg={result.get('message', '')}")
        return 0, []
    d = result.get("data") or {}
    cate_info = d.get("cateInfo") or {}
    cate_list = cate_info.get("categoryList") or []
    total = sum(c.get("num", 0) for c in cate_list)
    if not total:
        total = d.get("total", 0)
    return total, cate_list


def fetch_blog_list_page(qq_number, cookie_str, g_tk, pos, num=100):
    """获取一页日志列表"""
    params = {
        "inCharset": "utf-8",
        "outCharset": "utf-8",
        "format": "jsonp",
        "hostUin": qq_number,
        "uin": qq_number,
        "g_tk": str(g_tk),
        "blogType": "0",
        "reqInfo": "1",
        "pos": str(pos),
        "num": str(num),
    }
    return try_api_endpoints("/cgi-bin/blognew/get_abs", params, qq_number, cookie_str)


def fetch_blog_detail(qq_number, cookie_str, g_tk, blog_id):
    """获取单篇日志详情"""
    params = {
        "inCharset": "utf-8",
        "outCharset": "utf-8",
        "format": "jsonp",
        "uin": qq_number,
        "blogid": str(blog_id),
        "numperpage": "15",
        "ref": "qzone",
        "g_tk": str(g_tk),
    }
    result = try_api_endpoints("/cgi-bin/blognew/blog_output_data", params, qq_number, cookie_str)
    if not result:
        return ""
    # 如果是dict(JSONP解析成功)
    if isinstance(result, dict):
        d = result.get("data") or result
        if isinstance(d, dict):
            content = d.get("content") or d.get("html") or d.get("blogContent") or ""
            if content:
                return content
        elif isinstance(d, str):
            return d
        # 可能整个JSON里没有直接的content字段,尝试其他方式
        return str(result)
    # 如果是字符串(纯HTML响应)
    return result


def extract_blog_content(raw_text):
    """从HTML或JSONP响应中提取日志正文"""
    if not raw_text:
        return ""
    # 如果是dict
    if isinstance(raw_text, dict):
        d = raw_text.get("data") or raw_text
        if isinstance(d, dict):
            content = d.get("content") or d.get("html") or d.get("blogContent") or ""
            if content:
                return clean_html(content)
        elif isinstance(d, str):
            return clean_html(d)
        return ""
    # 字符串处理
    text = raw_text
    # 尝试提取 blogDetailDiv
    m = re.search(r'id=["\']blogDetailDiv["\'][^>]*>(.*?)</div>', text, re.DOTALL | re.IGNORECASE)
    if m:
        return clean_html(m.group(1))
    # 尝试提取 blogContainerDiv
    m = re.search(r'id=["\']blogContainerDiv["\'][^>]*>(.*?)</div>', text, re.DOTALL | re.IGNORECASE)
    if m:
        return clean_html(m.group(1))
    # 尝试提取 contentDiv
    m = re.search(r'class=["\']blog_detail["\'][^>]*>(.*?)</div>', text, re.DOTALL | re.IGNORECASE)
    if m:
        return clean_html(m.group(1))
    # 如果是完整HTML页面
    if "<html" in text.lower() or "<body" in text.lower():
        text = re.sub(r'<script[^>]*>.*?</script>', '', text, flags=re.DOTALL | re.IGNORECASE)
        text = re.sub(r'<style[^>]*>.*?</style>', '', text, flags=re.DOTALL | re.IGNORECASE)
        text = re.sub(r'<head[^>]*>.*?</head>', '', text, flags=re.DOTALL | re.IGNORECASE)
        return clean_html(text)
    # 直接清理
    return clean_html(text)


def blog_id_to_date(blog_id):
    """blogId 本质是Unix时间戳,转换为日期字符串"""
    try:
        ts = int(float(blog_id))
        if ts > 1000000000:
            return datetime.fromtimestamp(ts).strftime("%Y-%m-%d %H:%M:%S")
    except (ValueError, TypeError, OSError):
        pass
    return ""


def export_blogs(qq_number, cookie_str, g_tk, output_dir):
    """导出全部日志"""
    all_blogs = []
    pos = 0
    num = 100

    print()
    print("=" * 60)
    print("  开始导出日志")
    print("=" * 60)

    # 获取日志总数
    total, cate_list = fetch_blog_count(qq_number, cookie_str, g_tk)
    if cate_list:
        print(f"  日志分类:")
        for c in cate_list:
            print(f"    - {c.get('category', '未知')}: {c.get('num', 0)} 篇")
    print(f"  日志总数: {total} 篇")

    if total == 0:
        print("  [提示] 未获取到日志总数,尝试直接拉取...")

    while True:
        print(f"  获取日志列表 pos={pos} num={num} ...")
        data = fetch_blog_list_page(qq_number, cookie_str, g_tk, pos, num)

        if not data:
            print("  返回数据为空,停止")
            break

        if not isinstance(data, dict):
            print(f"  [错误] 返回的不是JSON,前200字符: {str(data)[:200]}")
            break

        code = data.get("code", -1)
        if code != 0:
            msg = data.get("message") or data.get("msg") or ""
            print(f"  接口返回 code={code}, message={msg}")
            print(f"  完整返回(前500字): {str(data)[:500]}")
            if code in [-3000, -3001, -4001]:
                print("  [提示] 登录态可能已失效,请重新获取Cookie!")
            break

        d = data.get("data") or {}
        blog_list = d.get("list") or d.get("blogList") or []

        if not blog_list:
            print("  没有更多日志了")
            break

        print(f"  本次获取到 {len(blog_list)} 篇日志")

        for blog in blog_list:
            blog_id = blog.get("blogId") or blog.get("blogid") or ""
            title = clean_html(blog.get("title", "无标题"))
            category = blog.get("cate") or blog.get("category") or "未分类"

            pub_time = blog.get("pubTime") or blog.get("pubtime") or ""
            if not pub_time and blog_id:
                pub_time = blog_id_to_date(blog_id)
            if not pub_time:
                pub_time = "未知日期"

            print(f"    获取详情: 《{title}》 ({pub_time})")

            content = ""
            if blog_id:
                raw = fetch_blog_detail(qq_number, cookie_str, g_tk, blog_id)
                if raw:
                    content = extract_blog_content(raw)
                else:
                    print(f"      [警告] 日志详情返回为空")
                time.sleep(0.5)

            all_blogs.append({
                "title": title,
                "date": pub_time,
                "category": category,
                "content": content,
                "blog_id": str(blog_id),
            })

        pos += num
        if total and pos >= total:
            print(f"  已获取全部 {total} 篇日志")
            break
        if len(blog_list) < num:
            print("  已获取全部日志")
            break

        time.sleep(REQUEST_DELAY)

    # ---- 写文件 ----
    blog_dir = os.path.join(output_dir, "日志")
    os.makedirs(blog_dir, exist_ok=True)

    for i, blog in enumerate(all_blogs, 1):
        safe_title = safe_filename(blog["title"])
        safe_date = blog["date"].replace(":", "-").replace(" ", "_") if blog["date"] else "未知日期"
        fname = f"{i:03d}_{safe_date}_{safe_title}.txt"
        fpath = os.path.join(blog_dir, fname)

        body = f"标题: {blog['title']}\n"
        body += f"日期: {blog['date']}\n"
        body += f"分类: {blog['category']}\n"
        body += f"日志ID: {blog.get('blog_id', '')}\n"
        body += "=" * 60 + "\n\n"
        body += blog["content"]

        save_file(fpath, body)

    summary_path = os.path.join(output_dir, "日志汇总.txt")
    lines = []
    lines.append("QQ空间日志导出")
    lines.append(f"QQ号: {qq_number}")
    lines.append(f"导出时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    lines.append(f"共计 {len(all_blogs)} 篇日志")
    lines.append("=" * 60)
    lines.append("")

    for blog in all_blogs:
        lines.append(f"标题: {blog['title']}")
        lines.append(f"日期: {blog['date']}")
        lines.append(f"分类: {blog['category']}")
        lines.append("-" * 60)
        lines.append(blog["content"])
        lines.append("=" * 60)
        lines.append("")

    save_file(summary_path, "\n".join(lines))

    print(f"\n  日志已保存:")
    print(f"    汇总文件: {summary_path}")
    print(f"    单篇目录: {blog_dir}")
    print(f"  共计 {len(all_blogs)} 篇")
    return len(all_blogs)


# ============================================================
#  主程序
# ============================================================

def main():
    print()
    print("=" * 60)
    print("  QQ空间 说说 + 日志 批量导出工具 v3")
    print("  仅使用Python标准库,无需安装任何依赖")
    print("=" * 60)
    print()

    qq_number = QQ_NUMBER.strip()
    cookie_str = COOKIE.strip()

    if not qq_number:
        qq_number = input("请输入你的QQ号: ").strip()
    else:
        print(f"QQ号: {qq_number}")

    if not cookie_str:
        print()
        print("需要完整Cookie才能访问QQ空间数据。")
        print()
        print("获取完整Cookie的方法(推荐用Network标签页):")
        print("  1. 浏览器打开 https://user.qzone.qq.com/%s 并登录" % qq_number)
        print("  2. 按 F12 打开开发者工具")
        print("  3. 切换到 Network(网络) 标签页")
        print("  4. 刷新页面(F5)")
        print("  5. 在请求列表中点击任意一个发往 qzone.qq.com 的请求")
        print("  6. 右侧 Headers(标头) → Request Headers → 找到 Cookie")
        print("  7. 复制完整的Cookie值(一整行很长的字符串)")
        print()
        print("  或者用Application标签页:")
        print("  F12 → Application → Cookies → 把所有项拼成 key=value; key=value 格式")
        print("  至少要包含 p_skey 和 uin 两个Cookie")
        print()
        cookie_str = input("请粘贴完整Cookie: ").strip()

    if not cookie_str:
        print("[错误] Cookie不能为空!")
        sys.exit(1)

    # 从Cookie中提取 p_skey
    p_skey = extract_cookie_value(cookie_str, "p_skey")
    if not p_skey:
        print("[错误] Cookie中未找到 p_skey,请确保复制了完整的Cookie!")
        sys.exit(1)

    g_tk = get_g_tk(p_skey)

    output_dir = OUTPUT_DIR
    os.makedirs(output_dir, exist_ok=True)

    print()
    print(f"QQ号: {qq_number}")
    print(f"p_skey: {p_skey[:10]}...{p_skey[-5:]}")
    print(f"g_tk: {g_tk}")
    print(f"Cookie长度: {len(cookie_str)} 字符")
    print(f"输出目录: {output_dir}")
    print()

    # 导出说说
    ss_count = export_shuoshuo(qq_number, cookie_str, g_tk, output_dir)

    # 导出日志
    blog_count = export_blogs(qq_number, cookie_str, g_tk, output_dir)

    # 总结
    print()
    print("=" * 60)
    print("  全部导出完成!")
    print(f"  说说: {ss_count} 条")
    print(f"  日志: {blog_count} 篇")
    print(f"  输出目录: {output_dir}")
    print("=" * 60)
    print()
    print("文件结构:")
    print(f"  {output_dir}/")
    print(f"    ├── 说说.txt          (全部说说汇总)")
    print(f"    ├── 日志汇总.txt       (全部日志汇总)")
    print(f"    └── 日志/              (每篇日志单独文件)")
    print(f"        ├── 001_日期_标题.txt")
    print(f"        ├── 002_日期_标题.txt")
    print(f"        └── ...")


if __name__ == "__main__":
    main()
最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
【社区内容提示】社区部分内容疑似由AI辅助生成,浏览时请结合常识与多方信息审慎甄别。
平台声明:文章内容(如有图片或视频亦包括在内)由作者上传并发布,文章内容仅代表作者本人观点,简书系信息发布平台,仅提供信息存储服务。

友情链接更多精彩内容