hiascend-forum-fetcher

name	hiascend-forum-fetcher
description	读取昇腾社区论坛（Hiascend Community Forum）的帖子内容，支持按时间范围筛选并导出为Excel。支持时间快捷选项（今年、本月、本周、今天）和自定义区间，自动剔除置顶贴（top>0），智能终止查询，时区自动转换（+8小时），并发分页获取提升效率，并将topicId转换为可点击的论坛链接。当用户需要获取昇腾论坛帖子、筛选特定时间范围的论坛内容、导出论坛数据到Excel时触发使用。

昇腾社区论坛数据获取工具

用于从昇腾社区论坛 API 获取帖子数据，按时间范围筛选，自动剔除置顶贴，支持并发分页获取，并导出为 Excel 格式。

功能特性

• 灵活的时间筛选：支持"今年"、"本月"、"本周"、"今天"快捷选项，或自定义日期区间，最小粒度为天
• 智能截止时间调整：若查询截止时间超过当天，自动调整为当天23:59:59
• 置顶贴自动剔除：自动过滤 resultList 中 top > 0 的置顶贴
• 智能查询终止：当当前批次最早创建时间早于查询区间时，自动停止后续查询
• 时区自动转换：createTime 自动加8小时（北京时间转换）
• 可配置分页大小：支持自定义每页获取数据量（默认100条）
• 并发分页获取：支持多线程并发获取，提升查询效率（默认6并发）
• Excel 导出：结构化输出，包含指定字段和可点击链接

API 端点

https://www.hiascend.com/ascendgateway/ascendservice/devCenter/bbs/servlet/get-topic-list?filterCondition=1&pageIndex={page}&pageSize={pageSize}

工作流程

1. 解析并调整时间范围

将用户输入转换为开始日期和结束日期（YYYYMMDDHHMMSS 格式）：

快捷选项：

• 今年 → 当年1月1日00:00:00 至 今天23:59:59
• 本月 → 当月1日00:00:00 至 今天23:59:59
• 本周 → 本周一00:00:00 至 今天23:59:59
• 今天 → 今天00:00:00 至 今天23:59:59

自定义区间：

• 格式：YYYY年MM月DD日 或 YYYY-MM-DD 或 YYYYMMDD
• 示例：2026年1月1号到3月30号 → start=20260101000000, end=20260330235959

截止时间调整：

• 若 end_date > today，则自动调整为 today 23:59:59
• 输出警告日志告知用户调整情况

2. 分页并发获取数据

两阶段获取策略：

阶段1：获取总页数

• 请求第1页，获取 totalCount
• 计算总页数：total_pages = ceil(totalCount / pageSize)

阶段2：并发获取

• 使用线程池并发获取剩余页面
• 默认并发数：6（可通过 --concurrency 调整）
• 每批次获取后处理数据（剔除置顶贴、时区转换、时间筛选）
• 智能终止：当某批次最早时间早于查询区间时，停止后续获取

3. 数据处理流程

步骤1：剔除置顶贴

• 过滤 resultList 中 top > 0 的数据
• 记录剔除数量

步骤2：时区转换

• 将 createTime（YYYYMMDDHHMMSS）加8小时
• 转换为北京时间

步骤3：时间筛选

• 筛选落在查询区间内的帖子
• 获取当前批次最早创建时间

步骤4：智能终止判断

• 若当前批次最早 createTime（加8小时后）< 查询区间开始时间
• 则停止后续查询，仅保留区间内数据

4. 输出格式（Excel）

将筛选后的数据保存为 Excel 文件（.xlsx），包含以下列：

列名	说明	格式/规则
topicId	帖子ID	resultList.topicId
sectionName	版块名称	resultList.sectionName
topicClassName	话题分类	resultList.topicClassName
title	帖子标题	resultList.title
createTime	创建时间	yyyy-mm-dd（已加8小时）
topicLink	帖子链接	https://www.hiascend.com/forum/thread-{topicId}-1-1.html

CLI 参数说明

python scripts/fetch_hiascend_forum.py <时间范围> [选项]

位置参数：

• time_range：时间范围文本，例如：今年、本月、2026年3月1日到3月30日

可选参数：

• --page-size N：每页获取数据量（默认 100）
• --concurrency N：并发数（默认 6，建议不超过10）
• --max-pages N：最大分页数（默认 500）
• --output <路径>：输出目录，默认当前工作目录
• --log-level <级别>：日志等级（DEBUG/INFO/WARNING/ERROR，默认 INFO）

实现脚本

#!/usr/bin/env python3
"""
昇腾社区论坛数据获取工具
支持：时间灵活筛选、置顶贴剔除、智能终止、时区转换、并发分页获取、Excel导出
"""

import argparse
import json
import logging
import math
import threading
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime, timedelta
import requests
import sys


def parse_date_range(time_input):
    """
    解析时间范围，支持快捷选项和自定义区间
    若截止时间超过当天，调整为当天23:59:59
    返回: (start_time, end_time) 格式 YYYYMMDDHHMMSS
    """
    today = datetime.now()
    today_end = today.replace(hour=23, minute=59, second=59)
    
    # 快捷选项
    if '今年' in time_input or '本年' in time_input:
        start_time = today.replace(month=1, day=1, hour=0, minute=0, second=0)
        end_time = today_end
    elif '本月' in time_input:
        start_time = today.replace(day=1, hour=0, minute=0, second=0)
        end_time = today_end
    elif '本周' in time_input:
        monday = today - timedelta(days=today.weekday())
        start_time = monday.replace(hour=0, minute=0, second=0)
        end_time = today_end
       time_input:
        start_time = today.replace(hour=, minute=, second=)
        end_time = today_end
    :
        
        dates = extract_dates(time_input)
         (dates) >= :
            start_dt = datetime.strptime(dates[], )
            end_dt = datetime.strptime(dates[], )
            start_time = start_dt.replace(hour=, minute=, second=)
            end_time = end_dt.replace(hour=, minute=, second=)
        :
             ValueError()
    
    
     end_time.date() > today.date():
        logging.warning(
                       )
        end_time = today_end
    
     start_time.strftime(), end_time.strftime()


 ():
    
     re
    dates = []
    patterns = [
        ,
        ,
        
    ]
     pattern  patterns:
        matches = re.findall(pattern, text)
           matches:
             (, ):
                y, m, d = 
                dates.append()
            :
                dates.append()
     dates[:]


 ():
    
     (create_time_str) >= :
        :
            dt = datetime.strptime(create_time_str[:], )
            dt = dt + timedelta(hours=)
             dt.strftime()
        :
            
     create_time_str


 ():
    
     (beijing_time_str) >= :
         
     beijing_time_str


 ():
    
    url = base_url.(page=page)
     attempt  (, retry_limit + ):
        :
            resp = requests.get(url, headers=headers, timeout=)
             resp.status_code != :
                 RuntimeError()
            data = resp.json()
             data.get() != :
                 RuntimeError()
             data.get(, {}).get(, [])  []
         Exception  e:
             attempt >= retry_limit:
                logging.warning()
                 []
            time.sleep((backoff ** attempt, ))
     []


 ():
    
    
    filtered = []
    pinned_count = 
     t  topics:
        top_value = t.get(, )
         top_value  (top_value) > :
            pinned_count += 
        :
            filtered.append(t)
    
      filtered:
         [], , pinned_count
    
    
    valid_topics = []
    earliest_beijing_time = 
    
     t  filtered:
        original_ct = (t.get(, ))
        beijing_ct = convert_to_beijing_time(original_ct)
        
        
         earliest_beijing_time    beijing_ct < earliest_beijing_time:
            earliest_beijing_time = beijing_ct
        
        
         start_time <= beijing_ct <= end_time:
            t[] = beijing_ct
            valid_topics.append(t)
    
     valid_topics, earliest_beijing_time, pinned_count


 ():
    
    base_url = (
                
                )
    headers = {
        : ,
        : ,
        : ,
    }
    
    all_topics = []
    total_pinned = 
    should_stop = 
    
    logging.info(
                )
    
    
    first_page_topics = fetch_page(, base_url, headers)
      first_page_topics:
        logging.error()
         []
    
    
    valid_topics, earliest_time, pinned = process_topics(first_page_topics, start_time, end_time)
    total_pinned += pinned
     valid_topics:
        all_topics.extend(valid_topics)
        logging.info(
                    )
    
    
     earliest_time  earliest_time < start_time:
        logging.info()
         all_topics
    
    
    :
        resp = requests.get(base_url.(page=), headers=headers, timeout=)
        data = resp.json()
        total_count = (data.get(, {}).get(, ))
        total_pages = math.ceil(total_count / page_size)
        total_pages = (total_pages, max_pages)
        logging.info()
     Exception  e:
        logging.warning()
        total_pages = max_pages
    
     total_pages <= :
         all_topics
    
    
    pages_to_fetch = ((, total_pages + ))
    mutex = threading.Lock()
    
     ThreadPoolExecutor(max_workers=concurrency)  executor:
        future_to_page = {
            executor.submit(fetch_page, p, base_url, headers): p 
             p  pages_to_fetch
        }
        
         future  as_completed(future_to_page):
             should_stop:
                
                 f  future_to_page:
                    f.cancel()
                
            
            page = future_to_page[future]
            :
                topics = future.result()
             Exception  e:
                logging.warning()
                
            
              topics:
                
            
            
            valid_topics, earliest_time, pinned = process_topics(topics, start_time, end_time)
            
             mutex:
                total_pinned += pinned
                 valid_topics:
                    all_topics.extend(valid_topics)
                
                
                 earliest_time  earliest_time < start_time:
                    logging.info(
                               )
                    should_stop = 
                :
                    logging.info(
                               )
    
     total_pinned > :
        logging.info()
    
     all_topics


 ():
    
    extracted = []
     t  topics:
        topic_id = t.get()  t.get()  
        beijing_time = t.get(, )
        
        row = {
            : topic_id,
            : t.get(, ),
            : t.get(, ),
            : t.get(, ),
            : format_date_display(beijing_time),  
            : 
        }
        extracted.append(row)
     extracted


 ():
    
    :
         pandas  pd
     ImportError:
        logging.error()
        sys.exit()
    
      filename:
        filename = 
    
    df = pd.DataFrame(data)
    cols = [, , , , , ]
    df = df[cols]
    
     pd.ExcelWriter(filename, engine=)  writer:
        df.to_excel(writer, sheet_name=, index=)
        ws = writer.sheets[]
        widths = [, , , , , ]
         i, w  (widths, start=):
            ws.column_dimensions[( + i)].width = w
    
    logging.info()


 ():
    parser = argparse.ArgumentParser(description=)
    parser.add_argument(, =)
    parser.add_argument(, =, default=, 
                       =)
    parser.add_argument(, =, default=,
                       =)
    parser.add_argument(, =, default=, =)
    parser.add_argument(, =)
    parser.add_argument(, default=, =)
    args = parser.parse_args()
    
    logging.basicConfig(
        level=(logging, args.log_level.upper(), logging.INFO),
        =
    )
    
    
    start_time, end_time = parse_date_range(args.time_range)
    logging.info()
    logging.info()
    
    
    topics = fetch_all_topics(start_time, end_time, args.page_size, args.concurrency, args.max_pages)
    logging.info()
    
    
    topics.sort(key= x: x.get(, ), reverse=)
    
    
    data = extract_required_fields(topics)
    
    
    filename = 
     args.output:
         os
        os.makedirs(args.output, exist_ok=)
        filename = os.path.join(args.output, 
                               )
    
    save_to_excel(data, start_time, end_time, filename)


 __name__ == :
    main()

依赖

• Python 3.7+
• requests
• pandas
• openpyxl

pip install requests pandas openpyxl

使用示例

# 快捷选项
python scripts/fetch_hiascend_forum.py "今年"
python scripts/fetch_hiascend_forum.py "本月"

# 自定义区间
python scripts/fetch_hiascend_forum.py "2026年1月1号到3月30号"

# 调整每页数据量和并发数
python scripts/fetch_hiascend_forum.py "今年" --page-size 50 --concurrency 8

# 指定输出目录
python scripts/fetch_hiascend_forum.py "今年" --output ./data