请求修改Python网页爬虫API：将案件ID范围改为日期范围

当前Python爬虫通过指定年份+案件ID范围的方式爬取法庭案件信息，需要修改为按日期范围爬取——指定某年份内的日期区间，遍历区间内每一天，获取当日所有案件。需调整以下3个Python文件：

1. 主爬虫文件（示例命名：main.py） #

原代码 #

import json
import os
import time

from progress.bar import Bar

from lib.giustizia import get_case_details
from lib.ranges import load_ids_from_json

SCAN_MODE = False
RATE_LIMIT = 0.2

INITIAL_ID = 49751
FINAL_ID = 49950
RANGE_YEAR = 2021

query_range = {RANGE_YEAR: range(INITIAL_ID, FINAL_ID, 1)}

if os.path.exists("json_results.txt") and not SCAN_MODE:
    loaded_json = load_ids_from_json("json_results.txt")
    query_range = loaded_json if loaded_json else query_range
"""
json_results = open("json_results.txt", "w+")
csv_results = open("csv_results.csv", "w+")
"""
html_results = open("html_results.html", "w+")
for year in query_range:
    print("Querying cases from year {}".format(year))

    for process_id in Bar('Querying', suffix='%(percent).1f%% - %(eta)ds').iter(query_range[year]):
        case = get_case_details(year, process_id)

        if case:
            print(" - {}".format(case))
            """
            json_results.write(json.dumps(case.asdict()))
            csv_results.write(str(case))
            """
            html_results.write(str(case))
            """
            json_results.write("\n")
            csv_results.write("\n")
            """
            html_results.write("\n")
            """
            json_results.flush()
            csv_results.flush()
            """
            html_results.flush()
        else:
            print(process_id, "errored")

        time.sleep(RATE_LIMIT)  # wait a little to prevent DOS

"""
json_results.close()
csv_results.close()
"""
html_results.close()

修改后代码 #

import json
import os
import time
from datetime import datetime, timedelta

from progress.bar import Bar

from lib.giustizia import get_cases_by_date
from lib.ranges import load_crawled_dates

SCAN_MODE = False
RATE_LIMIT = 0.2

# 配置日期范围：格式YYYY/MM/DD
START_DATE = "2022/08/08"
END_DATE = "2022/08/29"

# 生成日期区间内的所有日期
def generate_date_range(start_str, end_str):
    start_date = datetime.strptime(start_str, "%Y/%m/%d")
    end_date = datetime.strptime(end_str, "%Y/%m/%d")
    date_range = []
    current_date = start_date
    while current_date <= end_date:
        date_range.append(current_date.strftime("%Y/%m/%d"))
        current_date += timedelta(days=1)
    return date_range

query_dates = generate_date_range(START_DATE, END_DATE)

# 加载已爬取的日期，避免重复爬取
if os.path.exists("crawled_dates.txt") and not SCAN_MODE:
    crawled_dates = load_crawled_dates("crawled_dates.txt")
    query_dates = [date for date in query_dates if date not in crawled_dates]

html_results = open("html_results.html", "a+")  # 改为追加模式避免覆盖
crawled_dates_file = open("crawled_dates.txt", "a+")

for date in Bar('Querying dates', suffix='%(percent).1f%% - %(eta)ds').iter(query_dates):
    print(f"Querying cases from date {date}")
    cases = get_cases_by_date(date)
    
    if cases:
        for case in cases:
            print(f" - {case}")
            html_results.write(str(case))
            html_results.write("\n")
            html_results.flush()
    else:
        print(f"{date}: no cases or error occurred")
    
    # 记录已爬取的日期
    crawled_dates_file.write(date + "\n")
    crawled_dates_file.flush()
    time.sleep(RATE_LIMIT)

html_results.close()
crawled_dates_file.close()

修改说明 #

• 替换ID范围配置为日期范围配置，新增generate_date_range函数生成区间内所有日期
• 替换原单案件查询函数为按日期批量查询的get_cases_by_date
• 修改已爬取记录逻辑：从记录案件ID改为记录已爬取日期，避免重复爬取
• 结果文件改为追加模式，防止覆盖历史数据

2. 依赖文件：lib/giustizia.py #

原代码片段 #

# -*- coding: utf-8 -*-
import re
from datetime import datetime

import requests
from bs4 import BeautifulSoup

from secrets import (
    deviceheight, devicename, devicewidth, os_version, token,
    user_agent, uuid
)

QUERY_URL = "https://mob1.processotelematico.giustizia.it/proxy/index_mobile.php"
BASE_PAYLOAD = dict(
    version="2.0.15",
    platform=os_version,
    uuid=uuid,
    devicename=devicename,
    devicewidth=devicewidth,
    deviceheight=deviceheight,
    token=token,
    azione="direttarg_sicid_mobile",
    registro="CC",
    idufficio="0580910098",
    tipoufficio=1
)
HEADERS = {
    'User-Agent': user_agent
}

RE_INSCRITO_RUOLO = re.compile("\<li\>iscritto al ruolo il (.+)\<\/li\>")
RE_REMOVE_LAWYER_PREFIX = re.compile("(?<=Avv. ).*")


class Case:
    def __init__(self, case_yr, case_no, date_filed, judge_name, date_hearing, case_state, primary_lawyer_initials,
                 raw_case_content=None, judgement_number=None):
        self.year = case_yr
        self.number = case_no
        self.date_filed = date_filed
        self.date_hearing = date_hearing
        self.judge_name = judge_name
        self.case_state = case_state
        self.primary_lawyer_initials = primary_lawyer_initials
        self.raw_case_content = raw_case_content
        self.judgement_number = judgement_number

    def __str__(self):
        return ";".join([
            '{}/{}'.format(self.number, self.year),
            
            self.raw_case_content
            
        ])
        
     

    def asdict(self):
        return {
            
            'raw_case_content': self.raw_case_content,
   }


def get_case_details(case_yr, case_no):
    payload = BASE_PAYLOAD.copy()

    payload.update(dict(
        aaproc=str(case_yr),
        numproc=str(case_no),
        _=int(datetime.now().timestamp())
    ))

修改后代码（新增+修改部分） #

# -*- coding: utf-8 -*-
import re
from datetime import datetime

import requests
from bs4 import BeautifulSoup

from secrets import (
    deviceheight, devicename, devicewidth, os_version, token,
    user_agent, uuid
)

QUERY_URL = "https://mob1.processotelematico.giustizia.it/proxy/index_mobile.php"
BASE_PAYLOAD = dict(
    version="2.0.15",
    platform=os_version,
    uuid=uuid,
    devicename=devicename,
    devicewidth=devicewidth,
    deviceheight=deviceheight,
    token=token,
    azione="direttarg_sicid_mobile",
    registro="CC",
    idufficio="0580910098",
    tipoufficio=1
)
HEADERS = {
    'User-Agent': user_agent
}

RE_INSCRITO_RUOLO = re.compile(r"<li>iscritto al ruolo il (.+)</li>")
RE_REMOVE_LAWYER_PREFIX = re.compile(r"(?<=Avv. ).*")
# 新增：提取案件编号和年份的正则（根据实际返回内容调整）
RE_CASE_NUM_YEAR = re.compile(r"(\d+)/(\d+)")


class Case:
    def __init__(self, case_yr, case_no, date_filed, judge_name, date_hearing, case_state, primary_lawyer_initials,
                 raw_case_content=None, judgement_number=None):
        self.year = case_yr
        self.number = case_no
        self.date_filed = date_filed
        self.date_hearing = date_hearing
        self.judge_name = judge_name
        self.case_state = case_state
        self.primary_lawyer_initials = primary_lawyer_initials
        self.raw_case_content = raw_case_content
        self.judgement_number = judgement_number

    def __str__(self):
        return ";".join([
            f'{self.number}/{self.year}',
            self.raw_case_content or ""
        ])
        
    def asdict(self):
        return {
            'year': self.year,
            'number': self.number,
            'raw_case_content': self.raw_case_content,
            # 其他字段按需添加
        }


def get_case_details(case_yr, case_no):
    payload = BASE_PAYLOAD.copy()
    payload.update(dict(
        aaproc=str(case_yr),
        numproc=str(case_no),
        _=int(datetime.now().timestamp())
    ))
    # 原函数剩余逻辑（发送请求、解析Case对象）保留，此处省略


# 新增：按日期查询当日所有案件的函数
def get_cases_by_date(date_str):
    # 转换日期格式为接口要求的格式（假设接口需要DD/MM/YYYY，根据实际调整）
    date_obj = datetime.strptime(date_str, "%Y/%m/%d")
    formatted_date = date_obj.strftime("%d/%m/%Y")
    
    payload = BASE_PAYLOAD.copy()
    payload.update(dict(
        dal=formatted_date,
        al=formatted_date,  # 如果接口需要结束日期，和开始日期一致就是查询单日
        _=int(datetime.now().timestamp())
    ))
    
    try:
        response = requests.post(QUERY_URL, data=payload, headers=HEADERS)
        response.raise_for_status()
        soup = BeautifulSoup(response.text, "html.parser")
        
        cases = []
        # 解析页面中的案件列表（根据实际页面结构调整解析逻辑）
        case_elements = soup.find_all("div", class_="case-item")  # 示例选择器，需替换为实际元素
        
        for elem in case_elements:
            raw_content = elem.get_text(strip=True)
            # 从原始内容中提取案件编号和年份
            num_year_match = RE_CASE_NUM_YEAR.search(raw_content)
            if num_year_match:
                case_no = num_year_match.group(1)
                case_yr = num_year_match.group(2)
                # 其他字段（日期、法官等）按需从elem或raw_content中提取
                case = Case(
                    case_yr=case_yr,
                    case_no=case_no,
                    date_filed=date_str,
                    judge_name="",  # 需补充解析逻辑
                    date_hearing="",  # 需补充解析逻辑
                    case_state="",  # 需补充解析逻辑
                    primary_lawyer_initials="",  # 需补充解析逻辑
                    raw_case_content=raw_content
                )
                cases.append(case)
        return cases
    except Exception as e:
        print(f"Error fetching cases for {date_str}: {str(e)}")
        return None

修改说明 #

• 新增get_cases_by_date函数，使用dal（和al）参数构造日期查询的payload
• 补充日期格式转换逻辑，适配接口要求的日期格式
• 新增案件列表解析逻辑（需根据实际返回的HTML结构调整选择器和字段提取规则）
• 优化Case类的asdict方法，保留必要字段用于后续记录

3. 依赖文件：lib/ranges.py #

原代码 #

import json


def load_ids_from_json(filename):
    f = open(filename)
    json_objs = f.read().split("\n")
    f.close()

    sets = {}
    out = {}

    for oj in json_objs:
        try:
            o = json.loads(oj)
        except:
            continue

        if not sets.get(o['case_yr']):
            sets[o['case_yr']] = set()
        sets[o['case_yr']].add(int(o['case_no']))

    for year in sets:
        out[year] = list(sets[year])
        out[year].sort()

    return out

修改后代码 #

import os

def load_crawled_dates(filename):
    """加载已爬取的日期列表"""
    if not os.path.exists(filename):
        return set()
    
    with open(filename, "r") as f:
        dates = f.read().split("\n")
    # 过滤空行并转为集合
    return set(date.strip() for date in dates if date.strip())

修改说明 #

• 替换原load_ids_from_json为load_crawled_dates，改为读取已爬取的日期记录
• 使用集合存储日期，便于快速判断是否已爬取
• 简化文件读取逻辑，处理空行情况

内容的提问来源于stack exchange，提问作者user18918352