allowlgroup/parser
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

сделал ревью системы
All checks were successful
continuous-integration/drone/push Build is passing
Details

Browse Source
This commit is contained in:
Игорь Бандурист
2026-04-28 22:13:47 +10:00
parent c564140428
commit 25f2c09064
18 changed files with 1169 additions and 712 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
api/__init__.py Normal file
View File

@@ -0,0 +1,20 @@
"""
API модуль приложения
"""
from .schemas import (
ParserOneRequest,
Parserall,
Parserall_url,
Source,
DownloadRange
)
from .routes import setup_routes
__all__ = [
'ParserOneRequest',
'Parserall',
'Parserall_url',
'Source',
'DownloadRange',
'setup_routes'
]

258
api/routes.py Normal file
View File

@@ -0,0 +1,258 @@
"""
API эндпоинты приложения
"""
import os
import zipfile
from datetime import datetime, timedelta
from typing import List
from fastapi import BackgroundTasks, FastAPI, Query, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse
from config import DOCUMENTS_DIR, APP_TITLE, APP_DESCRIPTION, APP_VERSION
from utils import logger
from api.schemas import ParserOneRequest, Parserall, Source, DownloadRange
from parsers import start_pars_one_istochnik, start_pars_two_istochnik, start_pars_all_istochnik
import work_parser as wp
import parser_bd as pbd
def setup_routes(app: FastAPI) -> None:
"""
Настройка всех API маршрутов
"""
# CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# ==================== Парсеры ====================
@app.post("/parser_1", summary="Запуск процесса парсинга первого источника")
async def process_parser_one_ist(data: ParserOneRequest, background_tasks: BackgroundTasks):
istochnik = data.time.split("-")
background_tasks.add_task(start_pars_one_istochnik, istochnik)
return {"message": "Процесс парсинга 1 источника запущен"}
@app.post("/parser_2", summary="Запуск процесса парсинга второго источника")
async def process_parser_two_ist(background_tasks: BackgroundTasks):
background_tasks.add_task(start_pars_two_istochnik)
return {"message": "Процесс парсинга 2 источника запущен"}
@app.post("/add_sources", summary="Добавление парсинга любого источника")
async def add_sources_all_ist(sources: Parserall):
result = wp.add_sources(str(sources.url), sources.promt)
return {"status": "success", "message": "Источник добавлен", "data": result}
@app.get("/all_sources", summary="Метод получения всех источников")
async def get_all_sources():
return wp.get_all_sources()
@app.delete("/delete_sources", summary="Метод удаления источника")
async def delete_sources(url: str):
return print(wp.delete_sources(url))
@app.post("/parser_all", summary="Запуск процесса парсинга любого источника")
async def process_parser_all_ist(url: Parserall, background_tasks: BackgroundTasks):
background_tasks.add_task(start_pars_all_istochnik, str(url.url), url.promt)
return {"message": "Процесс парсинга любого источника запущен"}
@app.get("/get_tasks_offset", summary="Метод получения задач парсинга")
async def get_tasks_offset(limit: int = Query(10, gt=0), offset: int = Query(0, ge=0)):
return wp.get_tasks_offset(limit, offset)
# ==================== Настройки ====================
@app.get("/settings", summary="Метод получения настроек парсера")
async def get_settings():
return wp.get_all_promt()
@app.get("/categories_promt", summary="Метод получения categories_promt")
async def get_categories_promt():
return wp.get_all_categories_promt()
@app.post("/settings", summary="Метод сохранения настроек парсера")
async def set_settings(settings: Source):
return wp.update_promt(settings.name, settings.promt)
# ==================== Задачи ====================
@app.delete("/delete_task/{task_id}", summary="Метод удаления задачи")
async def delete_task(task_id: int):
return print(wp.delete_task(task_id))
# ==================== Файлы ====================
@app.get("/file_download", summary="Метод для скачивания файла")
async def download_file(path: str, title: str):
file_name = f"{title}.docx"
file_path = os.path.join(DOCUMENTS_DIR, path, file_name)
logger.warning(f"Файл: {file_path}")
if not os.path.exists(file_path):
logger.warning(f"Файл не найден: {file_path}")
return {"error": "Файл не найден", "path": file_path}
response = FileResponse(
path=file_path,
filename=file_name,
media_type="application/vnd.openxmlformats-officedocument.wordprocessingml.document"
)
response.headers["Access-Control-Allow-Origin"] = "*"
response.headers["Access-Control-Allow-Methods"] = "GET, OPTIONS"
response.headers["Access-Control-Allow-Headers"] = "Content-Type"
logger.warning(response)
return response
@app.post("/download_all", summary="Скачать все файлы за период")
async def download_all(dates: DownloadRange, background_tasks: BackgroundTasks):
date_start = dates.data_start
date_finish = dates.data_finish
try:
start_date = datetime.strptime(date_start, "%Y-%m-%d")
finish_date = datetime.strptime(date_finish, "%Y-%m-%d")
except ValueError:
return {"error": "Неверный формат даты. Используйте YYYY-MM-DD"}
if start_date > finish_date:
return {"error": "Дата начала не может быть позже даты окончания"}
all_files = []
current_date = start_date
while current_date <= finish_date:
date_path = current_date.strftime("%Y/%m/%d")
full_dir_path = os.path.join(DOCUMENTS_DIR, date_path)
if os.path.exists(full_dir_path):
for file in os.listdir(full_dir_path):
if file.endswith('.docx'):
all_files.append(os.path.join(full_dir_path, file))
current_date += timedelta(days=1)
if not all_files:
return {"error": "Файлы не найдены за указанный период", "date_start": date_start, "date_finish": date_finish}
archive_name = f"documents_{date_start}_{date_finish}.zip"
archive_path = os.path.join(DOCUMENTS_DIR, archive_name)
try:
with zipfile.ZipFile(archive_path, 'w', zipfile.ZIP_DEFLATED) as zipf:
for file_path in all_files:
zipf.write(file_path, os.path.basename(file_path))
except Exception as e:
logger.error(f"Ошибка создания архива: {e}")
return {"error": f"Ошибка создания архива: {e}"}
def cleanup_archive():
try:
if os.path.exists(archive_path):
os.remove(archive_path)
logger.info(f"Архив удалён: {archive_path}")
except Exception as e:
logger.warning(f"Не удалось удалить архив: {e}")
response = FileResponse(
path=archive_path,
filename=archive_name,
media_type="application/zip"
)
response.headers["Access-Control-Allow-Origin"] = "*"
response.headers["Access-Control-Allow-Methods"] = "GET, POST, OPTIONS"
response.headers["Access-Control-Allow-Headers"] = "Content-Type, Authorization"
response.headers["Access-Control-Expose-Headers"] = "Content-Disposition"
background_tasks.add_task(cleanup_archive)
return response
@app.get("/logs", summary="Показать логи")
async def get_logs():
with open("app.log", "r") as file:
lines = file.readlines()[-10:]
return {"logs": lines}
# ==================== Эндпоинты из parser_bd.py ====================
@app.post("/save_parsed_data", summary="Сохранить данные парсинга")
def save_parsed_data(data: pbd.ParsedData):
try:
pbd.save_parsed_data_to_db(data)
return {"status": "success", "message": "Данные успешно сохранены"}
except Exception as e:
raise HTTPException(status_code=500, detail=f"Ошибка при сохранении данных: {e}")
@app.post("/update_viewed_status", summary="Обновляет поле viewed")
def update_viewed_status(url: str, viewed: bool):
try:
result = pbd.update_viewed_status_in_db(url, viewed)
if not result.get("found"):
raise HTTPException(status_code=404, detail="Запись с указанным URL не найдена")
except Exception as e:
if isinstance(e, HTTPException):
raise e
raise HTTPException(status_code=500, detail=f"Ошибка при сохранении данных: {e}")
return {"status": "success", "message": "Статус просмотра успешно обновлен"}
@app.post("/update_status_status", summary="Обновляет поле status")
def update_status_status(url: str, status: bool):
try:
result = pbd.update_status_status_in_db(url, status)
if not result.get("found"):
raise HTTPException(status_code=404, detail="Запись с указанным URL не найдена")
except Exception as e:
if isinstance(e, HTTPException):
raise e
raise HTTPException(status_code=500, detail=f"Ошибка при сохранении данных: {e}")
return {"status": "success", "message": "Статус просмотра успешно обновлен"}
@app.get("/check_url_exists", summary="Проверяет url")
def check_url_exists(url: str):
try:
return pbd.check_url_exists_in_db(url)
except Exception as e:
raise HTTPException(status_code=500, detail=f"Ошибка при проверке: {e}")
@app.get("/records", summary="Получить записи из БД с пагинацией", response_model=List[pbd.ParsedData])
def get_records(offset: int = Query(0, ge=0), limit: int = Query(10, ge=1, le=100)):
try:
return pbd.get_records_from_db(offset, limit)
except Exception as e:
raise HTTPException(status_code=500, detail=f"Ошибка при получении записей из БД: {e}")
@app.get("/records_all/count", summary="Получить общее количество записей")
def get_records_count(item: str = Query("default")):
try:
return pbd.get_records_count_from_db(item)
except Exception as e:
raise HTTPException(status_code=500, detail=f"Ошибка при получении количества: {e}")
@app.get("/poisk/count", summary="Получить количество результатов поиска")
def get_poisk_count(query: str, item: str = Query("default")):
try:
return pbd.get_poisk_count_from_db(query, item)
except Exception as e:
raise HTTPException(status_code=500, detail=f"Ошибка при получении количества: {e}")
@app.get("/poisk", summary="Поиск с пагинацией")
def poisk(query: str, offset: int = Query(0, ge=0), limit: int = Query(10, ge=1, le=100), item: str = Query("default")):
try:
return pbd.poisk_from_db(query, offset, limit, item)
except Exception as e:
raise HTTPException(status_code=500, detail=f"Ошибка при поиске: {e}")
@app.get("/records_all", summary="Получить все записи из БД + сортирует + пагинация", response_model=List[pbd.ParsedData])
def get_records_all(item: str = Query("default"), offset: int = Query(0, ge=0), limit: int = Query(10, ge=1, le=100)):
try:
return pbd.get_records_all_from_db(item, offset, limit)
except Exception as e:
raise HTTPException(status_code=500, detail=f"Ошибка при получении записей из БД: {e}")

33
api/schemas.py Normal file
View File

@@ -0,0 +1,33 @@
"""
Pydantic схемы для API
"""
from pydantic import BaseModel, HttpUrl
from typing import List
class ParserOneRequest(BaseModel):
"""Запрос для парсинга первого источника"""
time: str
class Parserall(BaseModel):
"""Запрос для парсинга любого источника"""
url: HttpUrl
promt: str
class Parserall_url(BaseModel):
"""Запрос URL для источника"""
url: HttpUrl
class Source(BaseModel):
"""Модель источника для настроек"""
name: str
promt: str
class DownloadRange(BaseModel):
"""Диапазон дат для скачивания файлов"""
data_start: str
data_finish: str

36
config.py Normal file
View File

@@ -0,0 +1,36 @@
"""
Конфигурация приложения
"""
import os
# Пути
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
DOCUMENTS_DIR = os.path.join(BASE_DIR, "documents")
LOG_FILE = os.path.join(BASE_DIR, "app.log")
# GPT сервер
GPT_SERVER_URL = os.getenv('GPT_SERVER_URL', 'http://45.129.78.228:8484')
# GPT_SERVER_URL = os.getenv('GPT_SERVER_URL', 'http://127.0.0.1:8484')
# Прокси
PROXIES_URL = "https://raw.githubusercontent.com/TheSpeedX/SOCKS-List/master/http.txt"
# FastAPI настройки
APP_TITLE = "Parser API"
APP_DESCRIPTION = "API для запуска парсинга в базу данных"
APP_VERSION = "1.0"
UVICORN_PORT = 8001
# Настройки парсера
PARSER_TIMEOUT = 10
GPT_TIMEOUT = 60
GPT_MAX_RETRIES = 5
MAX_ARTICLE_TEXT_LENGTH = 4500
MIN_ARTICLE_TEXT_LENGTH = 100
MIN_UNIVERSAL_ARTICLE_TEXT_LENGTH = 200
# Планировщик
SCHEDULED_PARSER_1_HOUR = 0
SCHEDULED_PARSER_1_MINUTE = 0
SCHEDULED_PARSER_2_HOUR = 1
SCHEDULED_PARSER_2_MINUTE = 0

743
main.py
View File

@@ -1,735 +1,54 @@
# Стандартные библиотеки (stdlib)
import json
import logging
import os
import subprocess
import time
from datetime import datetime as dt
from datetime import datetime, timedelta
import random
import zipfile
import tempfile
# Сторонние библиотеки (third-party)
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from bs4 import BeautifulSoup
"""
Parser API - Точка входа приложения
"""
from contextlib import asynccontextmanager
from docx import Document
from newspaper import Article
from fastapi import BackgroundTasks, FastAPI, Query, Request, Depends, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse
from pydantic import BaseModel, HttpUrl
from typing import List
from urllib.parse import urljoin, urlparse, urldefrag
from fastapi import FastAPI
import uvicorn
import requests
# Локальные импорты
import parser_bd as pbd
import work_parser as wp
from config import (
APP_TITLE,
APP_DESCRIPTION,
APP_VERSION,
UVICORN_PORT,
SCHEDULED_PARSER_1_HOUR,
SCHEDULED_PARSER_1_MINUTE,
SCHEDULED_PARSER_2_HOUR,
SCHEDULED_PARSER_2_MINUTE
)
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from parsers import scheduled_parser_1, scheduled_parser_2
from api import setup_routes
# Инициализация планировщика
scheduler = AsyncIOScheduler()
DOCUMENTS_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "documents")
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Управление жизненным циклом приложения"""
# Startup
scheduler.add_job(scheduled_parser_1, "cron", hour=0, minute=0)
scheduler.add_job(scheduled_parser_2, "cron", hour=1, minute=0)
scheduler.add_job(scheduled_parser_1, "cron", hour=SCHEDULED_PARSER_1_HOUR, minute=SCHEDULED_PARSER_1_MINUTE)
scheduler.add_job(scheduled_parser_2, "cron", hour=SCHEDULED_PARSER_2_HOUR, minute=SCHEDULED_PARSER_2_MINUTE)
scheduler.start()
yield
# Shutdown
scheduler.shutdown()
app = FastAPI(title="Parser API",
description="API для запуска парсинга в базу данных",
version="1.0",
lifespan=lifespan)
# Инициализация планировщика
scheduler = AsyncIOScheduler()
# Настройка логгера
logging.basicConfig(filename="app.log", level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)
# Инициализация таблицы статуса парсинга
# wp.create_table()
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # или список разрешенных адресов, например ["https://allowlgroup.ru","http://localhost:5173", "http://45.129.78.228:8000"]
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
# expose_headers=["*"],
# Создание приложения FastAPI
app = FastAPI(
title=APP_TITLE,
description=APP_DESCRIPTION,
version=APP_VERSION,
lifespan=lifespan
)
PROXIES_URL = "https://raw.githubusercontent.com/TheSpeedX/SOCKS-List/master/http.txt"
def download_proxies(url):
response = requests.get(url)
if response.status_code == 200:
proxies = response.text.splitlines() # список прокси по строкам
return proxies
else:
return []
def fetch_with_proxy(url, proxy, verify, timeout):
proxies = {
'http': f'http://{proxy}',
'https': f'http://{proxy}',
}
try:
response = requests.get(url, proxies=proxies, timeout=timeout, verify=verify)
response.encoding = 'utf-8'
if response.status_code == 200:
# Проверяем содержимое - если это ошибка от прокси
if '"message":"Request failed' in response.text or '403' in response.text[:500]:
print(f"Proxy {proxy} - Site returned 403 (inside response)")
return None
print(f"Proxy {proxy} - SUCCESS")
return response.text
elif response.status_code == 403:
print(f"Proxy {proxy} - 403 Forbidden")
return None # Прокси работает, но сайт блокирует
else:
print(f"Proxy {proxy} - Status {response.status_code}")
return None
except:
return None
# Перемешивает список прокси для случайного начала
def get_shuffled_proxies(proxies_list):
shuffled = proxies_list.copy()
random.shuffle(shuffled)
return shuffled
# Общие функции нахождения ссылок
def extract_map_area_hrefs(url, verify=True, ist_number=1):
headers = {
"User-Agent": "Mozilla/5.0 (compatible; MyScraper/1.0; +https://example.com)"
}
resp = requests.get(url, headers=headers, verify=verify)
resp.raise_for_status()
soup = BeautifulSoup(resp.text, "html.parser")
hrefs = []
if ist_number == 1:
for map_tag in soup.find_all("li", attrs={"data-page": "1"}):
for a in map_tag.find_all("a", href=True):
href = a["href"]
abs_url = urljoin(url, href)
print(abs_url)
hrefs.append(abs_url)
else:
for map_tag in soup.find_all("map"):
for area in map_tag.find_all("area", href=True):
href = area["href"]
abs_url = urljoin(url, href)
hrefs.append(abs_url)
return hrefs
# функции парсера первого источника (газета)
def extract_text_from_url_one(url, timeout=10, verify=True):
proxies_list = download_proxies(PROXIES_URL)
proxies_list = get_shuffled_proxies(proxies_list)
response = ""
for proxy in proxies_list:
response = fetch_with_proxy(url, proxy=proxy, timeout=timeout, verify=verify)
if response:
break
else:
response = ""
soup = BeautifulSoup(response, "html.parser")
title_div = soup.find('div', class_='newsdetatit')
title_text = ''
if title_div:
h3_tag = title_div.find('h3')
if h3_tag:
title_text = h3_tag.get_text(strip=True)
content_div = soup.find('div', class_='newsdetatext')
content_text = ''
if content_div:
founder_content = content_div.find('founder-content')
if founder_content:
p_tags = founder_content.find_all('p')
content_text = '\n'.join(p.get_text(strip=True) for p in p_tags)
text = title_text + content_text
if len(text) > 4500:
text = text[:4500]
print(len(text))
return text
#Функции парсера второго источника (военного)
def extract_text_from_url(url, timeout=10, verify=True):
proxies_list = download_proxies(PROXIES_URL)
proxies_list = get_shuffled_proxies(proxies_list)
response = ""
for proxy in proxies_list:
response = fetch_with_proxy(url, proxy=proxy, timeout=timeout, verify=verify)
if response:
break
else:
response = ""
soup = BeautifulSoup(response, 'html.parser')
# Находим контейнер div.whitecon.article
container = soup.find("div", class_="whitecon article")
if not container:
return "", ""
# Получение заголовка <time> внутри контейнера
time_text = container.find('span')
if time_text:
time_t= time_text.get_text(strip=True)
# Получение всех <p> внутри контейнера, исключая те с class="before_ir"
paragraphs = container.find_all('p')
# Возвращаем текстовую сводку
content_text = []
for p in paragraphs:
if p.get('class') != ['before_ir'] :
content_text.append(p.get_text(strip=True))
return "\n".join(content_text), time_t
# GPT_SERVER_URL = os.getenv('GPT_SERVER_URL', 'http://45.129.78.228:8484')
# GPT_SERVER_URL = os.getenv('GPT_SERVER_URL', 'http://172.17.0.1:8484')
GPT_SERVER_URL = os.getenv('GPT_SERVER_URL', 'http://127.0.0.1:8484')
def gpt_response_message(content: str, name_promt: str):
contentGPT = wp.get_promt(name_promt).replace('{content}', content)
url = GPT_SERVER_URL
params = {'text': contentGPT}
max_retries = 5
retries = 0
while retries < max_retries:
try:
response = requests.get(url, params=params, timeout=60)
return response.text
except requests.exceptions.ConnectTimeout as e:
print(f"Ошибка подключения (timeout): {e}")
logger.warning(f"gpt_response_message timeout:") #{e}")
retries += 1
if retries < max_retries:
time.sleep(2 ** (retries - 1))
except requests.exceptions.ConnectionError as e:
print(f"Ошибка соединения: {e}")
logger.warning(f"gpt_response_message connection error: ") #{e}")
retries += 1
if retries < max_retries:
time.sleep(2 ** (retries - 1))
except Exception as ex:
print(f"Ошибка при запросе к GPT: {ex}")
logger.error(f"gpt_response_message: ") #{ex}")
retries += 1
if retries < max_retries:
time.sleep(2 ** (retries - 1))
logger.info(f"Привышен лимит запросов {max_retries}")
return ""
# Общие функции проверки ссылок
def check_url(url):
try:
result = pbd.check_url_exists_in_db(url)
return result.get("exists", False)
except Exception as e:
# Если ошибка — считаем, что URL новый (пропускаем)
logger.error(f"check_url error: {e}")
return False
# функции даты первого источника (газета)
def create_folder(num):
if int(num) // 10 == 0:
num = f"0{num}"
else:
num = str(num)
return num
# Функция формирования документа
def update_bd_and_create_document(response_text, article_date, url, parsed_at, original_text, other):
clean_response = ''
if not response_text:
print(f"Пустой ответ от GPT для URL: {url}")
logger.info(f"Пустой ответ от GPT для URL: {url}")
return
try:
clean_response = response_text.strip().replace('```json', '').replace('```', '').strip()
data = json.loads(clean_response)
if data['category']:
data['article_date'] = article_date
data['url'] = url
data['parsed_at'] = parsed_at
data['original_text'] = original_text
data['status'] = False
data['viewed'] = False
data['other'] = other
# Заменяем HTTP-запрос на прямой вызов функции через pbd
parsed_data = pbd.ParsedData(**data)
pbd.save_parsed_data_to_db(parsed_data)
print("Данные успешно сохранены в БД")
path_day = article_date.split()[0]
documents_path = os.path.join(DOCUMENTS_DIR, path_day)
if not os.path.exists(documents_path):
os.makedirs(documents_path)
print(f"Создана папка: {documents_path}")
doc = Document()
doc.add_heading('Ссылка на статью', level=1)
doc.add_paragraph(other)
doc.add_heading('Дата и время', level=1)
doc.add_paragraph(article_date)
doc.add_heading('Обноруженные тематики текста', level=1)
doc.add_paragraph(data["category"])
doc.add_heading('Заголовок', level=1)
doc.add_paragraph(data["title"])
doc.add_heading('Краткий пересказ', level=1)
doc.add_paragraph(data["short_text"])
doc.add_heading('Переведенный текст статьи в газете', level=1)
doc.add_paragraph(data["translation_text"])
doc.add_heading('Оригинальный текст', level=1)
doc.add_paragraph(original_text)
doc_name = f"{data['title']}.docx"
doc_path = os.path.join(documents_path, doc_name)
doc.save(doc_path)
print(f"Сохранен документ: {doc_path}")
except Exception as ex:
print(f"Ошибка при обработке ответа GPT: {ex}")
logger.info(f"Ошибка при обработке ответа GPT: {ex}")
#Функции start первого источника (газета)
def start_pars_one_istochnik(data_init=""):
if data_init != ['']:
current_day = data_init[2]
current_month = data_init[1]
current_year = data_init[0]
else:
datetime_now = dt.now()
current_day = create_folder(datetime_now.day)
current_month = create_folder(datetime_now.month)
current_year = f"{datetime_now.year}"
task_id = wp.insert_task(status='queued', source_url=f'http://epaper.hljnews.cn/hljrb/pc/layout/{current_year}{current_month}/{current_day}/node_0X.html')
print("Создана задача с id:", task_id)
for page_number in range(1, 9):
url = f'http://epaper.hljnews.cn/hljrb/pc/layout/{current_year}{current_month}/{current_day}/node_0{page_number}.html'
wp.update_task(task_id, status='in_progress', source_url=url, started_at=datetime.utcnow())
print(f"Сбор href из: {url}")
try:
hrefs = extract_map_area_hrefs(url, ist_number=2)
except Exception as e:
print(f"Ошибка при извлечении ссылок: {e}")
logger.info(f"extract_map_area_hrefs: {e}")
continue
for i, link in enumerate(hrefs, 1):
if check_url(link) == False:
print(f"Страница {page_number} [{i}/{len(hrefs)}] parsing {link}")
text = extract_text_from_url_one(link)
if len(text) >= 100:
response_text = gpt_response_message(text, "source1")
print(response_text)
if response_text:
update_bd_and_create_document(response_text=response_text, article_date=f"{current_year}/{current_month}/{current_day}", url=link, parsed_at=str(dt.now()), original_text=text, other="source1")
wp.update_task(task_id, status='completed', finished_at=datetime.utcnow())
#Функции start второго источника (военного)
def start_pars_two_istochnik():
task_id = wp.insert_task(status='queued', source_url=f'https://def.ltn.com.tw/')
istochnik = ['https://def.ltn.com.tw/breakingnewslist', 'https://def.ltn.com.tw/list/11', 'https://def.ltn.com.tw/list/19', 'https://def.ltn.com.tw/list/17','https://def.ltn.com.tw/list/16']
all_links = []
for url in istochnik:
try:
print(f"Сбор href из: {url}")
all_links += extract_map_area_hrefs(url)
except Exception as e:
print(f"Ошибка при извлечении ссылок: {e}")
logger.info(f"Ошибка при извлечении ссылок: {e}")
continue
for hrefs in all_links:
if check_url(hrefs) == False:
try:
text, time_text = extract_text_from_url(hrefs)
if len(text) >= 100:
response_text = gpt_response_message(text, "source2")
print(response_text)
if response_text:
update_bd_and_create_document(response_text=response_text, article_date=time_text, url=hrefs, parsed_at=str(dt.now()), original_text=text, other="source2")
except:
continue
wp.update_task(task_id, status='completed', finished_at=datetime.utcnow())
#Функции start любого источника
def start_pars_all_istochnik(url:str, promt:str):
print(f"Начало парсинга: {url} с промтом: {promt}")
task_id = wp.insert_task(status='queued', source_url=url)
try:
response = requests.get(url)
response.raise_for_status()
except requests.RequestException:
return set()
soup = BeautifulSoup(response.text, 'html.parser')
base_domain = urlparse(url).netloc
# links = []
for a_tag in soup.find_all('a', href=True):
href = a_tag['href'].strip()
if not href or href.startswith('mailto:') or href.startswith('javascript:'):
continue
# Приведение к абсолютному URL и удаление якорей (#...)
abs_url = urljoin(url, href)
abs_url, _ = urldefrag(abs_url)
parsed = urlparse(abs_url)
# Фильтр: ссылка должна быть на тот же домен
if parsed.netloc != base_domain:
continue
# Фильтрация по ключевым словам (пример для новостных сайтов)
# path_lower = parsed.path.lower()
# if any(keyword in path_lower for keyword in ['/news/', 'article', '2026', '2027', '/blog/', '/post/']):
# print(f"Парсинг {abs_url}")
if check_url(abs_url) == False and wp.check_error_url(abs_url):
try:
article = Article(abs_url)
article.download()
article.parse()
print("URL:", abs_url)
if len(article.text) > 200 and article.publish_date:
time_text = article.publish_date.strftime("%Y/%m/%d %H:%M:%S")
# print("Заголовок:", article.title)
# print("Дата публикации:", time_text)
# print("Текст статьи:", article.text)
response_text = gpt_response_message(str(article.text), promt)
print(response_text)
if response_text:
update_bd_and_create_document(response_text=response_text, article_date=time_text, url=abs_url, parsed_at=str(dt.now()), original_text=article.text, other=promt)
else:
wp.add_error_url(url, abs_url)
except Exception as e:
print(f"Ошибка при обработке статьи {abs_url}: {e}")
logger.info(f"Ошибка при обработке статьи {abs_url}: {e}")
continue # Продолжаем со следующей статьей
wp.update_task(task_id, status='completed', finished_at=datetime.utcnow())
# start_pars_all_istochnik("https://www.asahi.com", "japan")
# Функции для автоматического запуска
def scheduled_parser_1():
start_pars_one_istochnik()
def scheduled_parser_2():
start_pars_two_istochnik()
class ParserOneRequest(BaseModel):
time: str
@app.post("/parser_1", summary="Запуск процесса парсинга первого источника")
async def process_parser_one_ist(data: ParserOneRequest, background_tasks: BackgroundTasks):
istochnik = data.time.split("-")
background_tasks.add_task(start_pars_one_istochnik, istochnik)
return {"message": "Процесс парсинга 1 источника запущен"}
@app.post("/parser_2" , summary="Запуск процеса парсинга второго источника")
async def process_parser_two_ist(background_tasks: BackgroundTasks):
background_tasks.add_task(start_pars_two_istochnik)
return {"message": "Процесс парсинга 2 источника запущен"}
class Parserall(BaseModel):
url: HttpUrl
promt: str
class Parserall_url(BaseModel):
url: HttpUrl
@app.post("/add_sources" , summary="Добавление парсинга любого источника")
async def add_sources_all_ist(sources: Parserall):
result = wp.add_sources(str(sources.url), sources.promt)
return {"status": "success", "message": "Источник добавлен", "data": result}
@app.get("/all_sources", summary="Метод получения всех источников")
async def get_all_sources():
return wp.get_all_sources()
@app.delete("/delete_sources", summary="Метод удаления источника")
async def delete_sources(url: str):
return print(wp.delete_sources(url))
@app.post("/parser_all" , summary="Запуск процеса парсинга любого источника")
async def process_parser_all_ist(url: Parserall, background_tasks: BackgroundTasks):
background_tasks.add_task(start_pars_all_istochnik, str(url.url), url.promt)
return {"message": "Процесс парсинга любого источника запущен"}
@app.get("/get_tasks_offset", summary="Метод получения задач парсинга")
async def get_tasks_offset(limit: int = Query(10, gt=0), offset: int = Query(0, ge=0)):
return wp.get_tasks_offset(limit, offset)
@app.get("/settings", summary="Метод получения настроек парсера")
async def get_settings():
return wp.get_all_promt()
@app.get("/categories_promt", summary="Метод получения categories_promt")
async def get_categories_promt():
return wp.get_all_categories_promt()
class Source(BaseModel):
name: str
promt: str
# POST метод для установки настроек
@app.post("/settings", summary="Метод сохранения настроек парсера")
async def set_settings(settings: Source):
return wp.update_promt(settings.name, settings.promt)
@app.delete("/delete_task/{task_id}", summary="Метод удаления задачи")
async def delete_task(task_id: int):
return print(wp.delete_task(task_id))
@app.get("/file_download", summary="Метод для скачивания файла")
async def download_file(path: str, title: str):
file_name = f"{title}.docx"
file_path = os.path.join(DOCUMENTS_DIR, path, file_name)
logger.warning(f"Файл: {file_path}")
# Проверяем существование файла
if not os.path.exists(file_path):
logger.warning(f"Файл не найден: {file_path}")
return {"error": "Файл не найден", "path": file_path}
# Возвращаем файл
response = FileResponse(
path=file_path,
filename=file_name,
media_type="application/vnd.openxmlformats-officedocument.wordprocessingml.document"
)
response.headers["Access-Control-Allow-Origin"] = "*"
response.headers["Access-Control-Allow-Methods"] = "GET, OPTIONS"
response.headers["Access-Control-Allow-Headers"] = "Content-Type"
logger.warning(response)
return response
class DownloadRange(BaseModel):
data_start: str
data_finish: str
@app.post("/download_all", summary="Скачать все файлы за период")
async def download_all(dates: DownloadRange, background_tasks: BackgroundTasks):
date_start = dates.data_start
date_finish = dates.data_finish
try:
start_date = datetime.strptime(date_start, "%Y-%m-%d")
finish_date = datetime.strptime(date_finish, "%Y-%m-%d")
except ValueError:
return {"error": "Неверный формат даты. Используйте YYYY-MM-DD"}
if start_date > finish_date:
return {"error": "Дата начала не может быть позже даты окончания"}
all_files = []
current_date = start_date
while current_date <= finish_date:
date_path = current_date.strftime("%Y/%m/%d")
full_dir_path = os.path.join(DOCUMENTS_DIR, date_path)
# logger.info(f"Проверяем путь: {full_dir_path}")
if os.path.exists(full_dir_path):
for file in os.listdir(full_dir_path):
if file.endswith('.docx'):
all_files.append(os.path.join(full_dir_path, file))
current_date += timedelta(days=1)
# logger.info(f"Найдено файлов: {len(all_files)}")
if not all_files:
return {"error": "Файлы не найдены за указанный период", "date_start": date_start, "date_finish": date_finish}
# Создаём архив
archive_name = f"documents_{date_start}_{date_finish}.zip"
archive_path = os.path.join(DOCUMENTS_DIR, archive_name)
try:
with zipfile.ZipFile(archive_path, 'w', zipfile.ZIP_DEFLATED) as zipf:
for file_path in all_files:
zipf.write(file_path, os.path.basename(file_path))
except Exception as e:
logger.error(f"Ошибка создания архива: {e}")
return {"error": f"Ошибка создания архива: {e}"}
# logger.info(f"Архив создан: {archive_path}")
# Функция для удаления архива после отдачи
def cleanup_archive():
try:
if os.path.exists(archive_path):
os.remove(archive_path)
logger.info(f"Архив удалён: {archive_path}")
except Exception as e:
logger.warning(f"Не удалось удалить архив: {e}")
# Возвращаем архив
response = FileResponse(
path=archive_path,
filename=archive_name,
media_type="application/zip"
)
response.headers["Access-Control-Allow-Origin"] = "*"
response.headers["Access-Control-Allow-Methods"] = "GET, POST, OPTIONS"
response.headers["Access-Control-Allow-Headers"] = "Content-Type, Authorization"
response.headers["Access-Control-Expose-Headers"] = "Content-Disposition"
# Удаляем архив после отправки через background task
background_tasks.add_task(cleanup_archive)
return response
@app.get("/logs", summary="Показать логи")
async def get_logs():
with open("app.log", "r") as file:
lines = file.readlines()[-10:] # последние 10 строк
return {"logs": lines}
# ==================== Эндпоинты из parser_bd.py ====================
@app.post("/save_parsed_data", summary="Сохранить данные парсинга")
def save_parsed_data(data: pbd.ParsedData):
try:
pbd.save_parsed_data_to_db(data)
return {"status": "success", "message": "Данные успешно сохранены"}
except Exception as e:
raise HTTPException(status_code=500, detail=f"Ошибка при сохранении данных: {e}")
@app.post("/update_viewed_status", summary="Обновляет поле viewed")
def update_viewed_status(url: str, viewed: bool):
"""
Обновляет поле 'viewed' записи в БД по заданному URL.
"""
try:
result = pbd.update_viewed_status_in_db(url, viewed)
if not result.get("found"):
raise HTTPException(status_code=404, detail="Запись с указанным URL не найдена")
except Exception as e:
if isinstance(e, HTTPException):
raise e
raise HTTPException(status_code=500, detail=f"Ошибка при сохранении данных: {e}")
return {"status": "success", "message": "Статус просмотра успешно обновлен"}
@app.post("/update_status_status", summary="Обновляет поле status")
def update_status_status(url: str, status: bool):
"""
Обновляет поле 'status' записи в БД по заданному URL.
"""
try:
result = pbd.update_status_status_in_db(url, status)
if not result.get("found"):
raise HTTPException(status_code=404, detail="Запись с указанным URL не найдена")
except Exception as e:
if isinstance(e, HTTPException):
raise e
raise HTTPException(status_code=500, detail=f"Ошибка при сохранении данных: {e}")
return {"status": "success", "message": "Статус просмотра успешно обновлен"}
@app.get("/check_url_exists", summary="Проверяет url")
def check_url_exists(url: str):
"""
Проверяет, есть ли указанный URL в базе данных.
Возвращает true, если есть, иначе false.
"""
try:
return pbd.check_url_exists_in_db(url)
except Exception as e:
raise HTTPException(status_code=500, detail=f"Ошибка при проверке: {e}")
@app.get("/records", summary="Получить записи из БД с пагинацией", response_model=List[pbd.ParsedData])
def get_records(offset: int = Query(0, ge=0), limit: int = Query(10, ge=1, le=100)):
"""
Возвращает записи из таблицы url с учетом offset и limit.
"""
try:
return pbd.get_records_from_db(offset, limit)
except Exception as e:
raise HTTPException(status_code=500, detail=f"Ошибка при получении записей из БД: {e}")
@app.get("/records_all/count", summary="Получить общее количество записей")
def get_records_count(item: str = Query("default")):
"""
Возвращает общее количество записей в таблице.
"""
try:
return pbd.get_records_count_from_db(item)
except Exception as e:
raise HTTPException(status_code=500, detail=f"Ошибка при получении количества: {e}")
@app.get("/poisk/count", summary="Получить количество результатов поиска")
def get_poisk_count(query: str, item: str = Query("default")):
try:
return pbd.get_poisk_count_from_db(query, item)
except Exception as e:
raise HTTPException(status_code=500, detail=f"Ошибка при получении количества: {e}")
@app.get("/poisk", summary="Поиск с пагинацией")
def poisk(query: str, offset: int = Query(0, ge=0), limit: int = Query(10, ge=1, le=100), item: str = Query("default")):
try:
return pbd.poisk_from_db(query, offset, limit, item)
except Exception as e:
raise HTTPException(status_code=500, detail=f"Ошибка при поиске: {e}")
@app.get("/records_all", summary="Получить все записи из БД + сортирует + пагинация", response_model=List[pbd.ParsedData])
def get_records_all(item: str = Query("default"), offset: int = Query(0, ge=0), limit: int = Query(10, ge=1, le=100)):
"""
Возвращает записи из таблицы url с учетом фильтрации и пагинации.
"""
try:
return pbd.get_records_all_from_db(item, offset, limit)
except Exception as e:
raise HTTPException(status_code=500, detail=f"Ошибка при получении записей из БД: {e}")
# Настройка маршрутов
setup_routes(app)
if __name__ == "__main__":
uvicorn.run("main:app", port=8001, reload=True)
uvicorn.run("main:app", port=UVICORN_PORT, reload=True)

9
models/__init__.py Normal file
View File

@@ -0,0 +1,9 @@
"""
Модели данных
Импортируем ParsedData из parser_bd для обратной совместимости
"""
import parser_bd as pbd
ParsedData = pbd.ParsedData
__all__ = ['ParsedData']

19
parsers/__init__.py Normal file
View File

@@ -0,0 +1,19 @@
"""
Парсеры приложения
"""
from .base import BaseParser
from .source1 import Source1Parser, start_pars_one_istochnik, scheduled_parser_1
from .source2 import Source2Parser, start_pars_two_istochnik, scheduled_parser_2
from .universal import UniversalParser, start_pars_all_istochnik
__all__ = [
'BaseParser',
'Source1Parser',
'start_pars_one_istochnik',
'scheduled_parser_1',
'Source2Parser',
'start_pars_two_istochnik',
'scheduled_parser_2',
'UniversalParser',
'start_pars_all_istochnik'
]

47
parsers/base.py Normal file
View File

@@ -0,0 +1,47 @@
"""
Базовый класс парсера
"""
from abc import ABC, abstractmethod
from typing import List
import work_parser as wp
class BaseParser(ABC):
"""
Базовый класс для всех парсеров
"""
def __init__(self, source_name: str):
self.source_name = source_name
self.task_id = None
def start_task(self, source_url: str) -> int:
"""
Создаёт задачу парсинга и возвращает её ID
"""
self.task_id = wp.insert_task(status='queued', source_url=source_url)
print(f"Создана задача с id: {self.task_id}")
return self.task_id
def complete_task(self) -> None:
"""
Завершает задачу парсинга
"""
if self.task_id:
from datetime import datetime
wp.update_task(self.task_id, status='completed', finished_at=datetime.utcnow())
def fail_task(self) -> None:
"""
Отмечает задачу как неудачную
"""
if self.task_id:
from datetime import datetime
wp.update_task(self.task_id, status='failed', finished_at=datetime.utcnow())
@abstractmethod
def parse(self) -> None:
"""
Основной метод парсинга - должен быть реализован в наследниках
"""
pass

161
parsers/source1.py Normal file
View File

@@ -0,0 +1,161 @@
"""
Парсер первого источника - газета (hljnews.cn)
"""
import requests
from bs4 import BeautifulSoup
from datetime import datetime
from urllib.parse import urljoin
from typing import List
from .base import BaseParser
from config import PARSER_TIMEOUT, MIN_ARTICLE_TEXT_LENGTH, MAX_ARTICLE_TEXT_LENGTH
from utils import logger, create_folder, get_current_date_parts
from services import fetch_with_proxy_retry, gpt_response_message, update_bd_and_create_document
import work_parser as wp
def extract_map_area_hrefs(url: str, verify: bool = True, ist_number: int = 1) -> List[str]:
"""
Извлекает ссылки из map/area тегов или li элементов
"""
headers = {
"User-Agent": "Mozilla/5.0 (compatible; MyScraper/1.0; +https://example.com)"
}
resp = requests.get(url, headers=headers, verify=verify)
resp.raise_for_status()
soup = BeautifulSoup(resp.text, "html.parser")
hrefs = []
if ist_number == 1:
for map_tag in soup.find_all("li", attrs={"data-page": "1"}):
for a in map_tag.find_all("a", href=True):
href = a["href"]
abs_url = urljoin(url, href)
print(abs_url)
hrefs.append(abs_url)
else:
for map_tag in soup.find_all("map"):
for area in map_tag.find_all("area", href=True):
href = area["href"]
abs_url = urljoin(url, href)
hrefs.append(abs_url)
return hrefs
def extract_text_from_url_one(url: str, timeout: int = PARSER_TIMEOUT, verify: bool = True) -> str:
"""
Извлекает текст из статьи первого источника (газета)
"""
response = fetch_with_proxy_retry(url, timeout=timeout, verify=verify)
soup = BeautifulSoup(response, "html.parser")
title_div = soup.find('div', class_='newsdetatit')
title_text = ''
if title_div:
h3_tag = title_div.find('h3')
if h3_tag:
title_text = h3_tag.get_text(strip=True)
content_div = soup.find('div', class_='newsdetatext')
content_text = ''
if content_div:
founder_content = content_div.find('founder-content')
if founder_content:
p_tags = founder_content.find_all('p')
content_text = '\n'.join(p.get_text(strip=True) for p in p_tags)
text = title_text + content_text
if len(text) > MAX_ARTICLE_TEXT_LENGTH:
text = text[:MAX_ARTICLE_TEXT_LENGTH]
print(len(text))
return text
def check_url(url: str) -> bool:
"""
Проверяет, существует ли URL в базе данных
"""
try:
response = wp.check_url_exists(url)
if response.status_code == 200:
result = response.json()
print(result["exists"])
return result["exists"]
else:
return False
except Exception:
return False
class Source1Parser(BaseParser):
"""
Парсер для первого источника - газета hljnews.cn
"""
def __init__(self):
super().__init__("source1")
def parse(self, data_init: str = "") -> None:
"""
Основной метод парсинга первого источника
"""
if data_init != ['']:
current_day = data_init[2]
current_month = data_init[1]
current_year = data_init[0]
else:
current_year, current_month, current_day = get_current_date_parts()
source_url = f'http://epaper.hljnews.cn/hljrb/pc/layout/{current_year}{current_month}/{current_day}/node_0X.html'
self.start_task(source_url)
for page_number in range(1, 9):
url = f'http://epaper.hljnews.cn/hljrb/pc/layout/{current_year}{current_month}/{current_day}/node_0{page_number}.html'
wp.update_task(self.task_id, status='in_progress', source_url=url, started_at=datetime.utcnow())
print(f"Сбор href из: {url}")
try:
hrefs = extract_map_area_hrefs(url, ist_number=2)
except Exception as e:
print(f"Ошибка при извлечении ссылок: {e}")
logger.info(f"extract_map_area_hrefs: {e}")
continue
for i, link in enumerate(hrefs, 1):
if not check_url(link):
print(f"Страница {page_number} [{i}/{len(hrefs)}] parsing {link}")
text = extract_text_from_url_one(link)
if len(text) >= MIN_ARTICLE_TEXT_LENGTH:
response_text = gpt_response_message(text, "source1")
print(response_text)
if response_text:
update_bd_and_create_document(
response_text=response_text,
article_date=f"{current_year}/{current_month}/{current_day}",
url=link,
parsed_at=str(datetime.now()),
original_text=text,
other="source1"
)
self.complete_task()
def start_pars_one_istochnik(data_init: str = "") -> None:
"""
Точка входа для парсинга первого источника
"""
parser = Source1Parser()
parser.parse(data_init)
def scheduled_parser_1() -> None:
"""
Функция для автоматического запуска по расписанию
"""
start_pars_one_istochnik()

161
parsers/source2.py Normal file
View File

@@ -0,0 +1,161 @@
"""
Парсер второго источника - военный (def.ltn.com.tw)
"""
import requests
from bs4 import BeautifulSoup
from datetime import datetime
from urllib.parse import urljoin
from typing import List, Tuple
from .base import BaseParser
from config import PARSER_TIMEOUT, MIN_ARTICLE_TEXT_LENGTH
from utils import logger
from services import fetch_with_proxy_retry, gpt_response_message, update_bd_and_create_document
import work_parser as wp
def extract_map_area_hrefs(url: str, verify: bool = True, ist_number: int = 1) -> List[str]:
"""
Извлекает ссылки из map/area тегов или li элементов
"""
headers = {
"User-Agent": "Mozilla/5.0 (compatible; MyScraper/1.0; +https://example.com)"
}
resp = requests.get(url, headers=headers, verify=verify)
resp.raise_for_status()
soup = BeautifulSoup(resp.text, "html.parser")
hrefs = []
if ist_number == 1:
for map_tag in soup.find_all("li", attrs={"data-page": "1"}):
for a in map_tag.find_all("a", href=True):
href = a["href"]
abs_url = urljoin(url, href)
print(abs_url)
hrefs.append(abs_url)
else:
for map_tag in soup.find_all("map"):
for area in map_tag.find_all("area", href=True):
href = area["href"]
abs_url = urljoin(url, href)
hrefs.append(abs_url)
return hrefs
def extract_text_from_url(url: str, timeout: int = PARSER_TIMEOUT, verify: bool = True) -> Tuple[str, str]:
"""
Извлекает текст и дату из статьи второго источника (военный)
Возвращает кортеж (текст, дата)
"""
response = fetch_with_proxy_retry(url, timeout=timeout, verify=verify)
soup = BeautifulSoup(response, 'html.parser')
# Находим контейнер div.whitecon.article
container = soup.find("div", class_="whitecon article")
if not container:
return "", ""
# Получение заголовка <time> внутри контейнера
time_text = container.find('span')
time_t = ""
if time_text:
time_t = time_text.get_text(strip=True)
# Получение всех <p> внутри контейнера, исключая те с class="before_ir"
paragraphs = container.find_all('p')
# Возвращаем текстовую сводку
content_text = []
for p in paragraphs:
if p.get('class') != ['before_ir']:
content_text.append(p.get_text(strip=True))
return "\n".join(content_text), time_t
def check_url(url: str) -> bool:
"""
Проверяет, существует ли URL в базе данных
"""
try:
response = wp.check_url_exists(url)
if response.status_code == 200:
result = response.json()
print(result["exists"])
return result["exists"]
else:
return False
except Exception:
return False
class Source2Parser(BaseParser):
"""
Парсер для второго источника - военный def.ltn.com.tw
"""
def __init__(self):
super().__init__("source2")
def parse(self) -> None:
"""
Основной метод парсинга второго источника
"""
self.start_task('https://def.ltn.com.tw/')
istochnik = [
'https://def.ltn.com.tw/breakingnewslist',
'https://def.ltn.com.tw/list/11',
'https://def.ltn.com.tw/list/19',
'https://def.ltn.com.tw/list/17',
'https://def.ltn.com.tw/list/16'
]
all_links = []
for url in istochnik:
try:
print(f"Сбор href из: {url}")
all_links += extract_map_area_hrefs(url)
except Exception as e:
print(f"Ошибка при извлечении ссылок: {e}")
logger.info(f"Ошибка при извлечении ссылок: {e}")
continue
for hrefs in all_links:
if not check_url(hrefs):
try:
text, time_text = extract_text_from_url(hrefs)
if len(text) >= MIN_ARTICLE_TEXT_LENGTH:
response_text = gpt_response_message(text, "source2")
print(response_text)
if response_text:
update_bd_and_create_document(
response_text=response_text,
article_date=time_text,
url=hrefs,
parsed_at=str(datetime.utcnow()),
original_text=text,
other="source2"
)
except:
continue
self.complete_task()
def start_pars_two_istochnik() -> None:
"""
Точка входа для парсинга второго источника
"""
parser = Source2Parser()
parser.parse()
def scheduled_parser_2() -> None:
"""
Функция для автоматического запуска по расписанию
"""
start_pars_two_istochnik()

114
parsers/universal.py Normal file
View File

@@ -0,0 +1,114 @@
"""
Парсер любого источника - универсальный парсер
"""
import requests
from bs4 import BeautifulSoup
from datetime import datetime
from newspaper import Article
from urllib.parse import urljoin, urlparse, urldefrag
from typing import Set
from .base import BaseParser
from utils import logger
from services import gpt_response_message, update_bd_and_create_document
import work_parser as wp
def check_url(url: str) -> bool:
"""
Проверяет, существует ли URL в базе данных
"""
try:
response = wp.check_url_exists(url)
if response.status_code == 200:
result = response.json()
print(result["exists"])
return result["exists"]
else:
return False
except Exception:
return False
class UniversalParser(BaseParser):
"""
Универсальный парсер для любого источника
"""
def __init__(self, url: str, promt: str):
super().__init__("universal")
self.url = url
self.promt = promt
def parse(self) -> None:
"""
Основной метод парсинга любого источника
"""
print(f"Начало парсинга: {self.url} с промтом: {self.promt}")
self.start_task(self.url)
try:
response = requests.get(self.url)
print(response.text)
response.raise_for_status()
except requests.RequestException:
print(f"Ошибка при запросе к {self.url}")
self.fail_task()
return
soup = BeautifulSoup(response.text, 'html.parser')
base_domain = urlparse(self.url).netloc
print(base_domain)
for a_tag in soup.find_all('a', href=True):
href = a_tag['href'].strip()
if not href or href.startswith('mailto:') or href.startswith('javascript:'):
continue
# Приведение к абсолютному URL и удаление якорей (#...)
abs_url = urljoin(self.url, href)
abs_url, _ = urldefrag(abs_url)
parsed = urlparse(abs_url)
# Фильтр: ссылка должна быть на тот же домен
if parsed.netloc != base_domain:
continue
print("URL:", abs_url)
if not check_url(abs_url) and wp.check_error_url(abs_url):
try:
article = Article(abs_url)
article.download()
article.parse()
if len(article.text) > 200 and article.publish_date:
time_text = article.publish_date.strftime("%Y/%m/%d %H:%M:%S")
response_text = gpt_response_message(str(article.text), self.promt)
print(response_text)
if response_text:
update_bd_and_create_document(
response_text=response_text,
article_date=time_text,
url=abs_url,
parsed_at=str(datetime.now()),
original_text=article.text,
other=self.promt
)
else:
wp.add_error_url(self.url, abs_url)
except Exception as e:
print(f"Ошибка при обработке статьи {abs_url}: {e}")
logger.info(f"Ошибка при обработке статьи {abs_url}: {e}")
continue
self.complete_task()
def start_pars_all_istochnik(url: str, promt: str) -> None:
"""
Точка входа для парсинга любого источника
"""
parser = UniversalParser(url, promt)
parser.parse()

20
services/__init__.py Normal file
View File

@@ -0,0 +1,20 @@
"""
Сервисы приложения
"""
from .proxy_manager import (
download_proxies,
get_shuffled_proxies,
fetch_with_proxy,
fetch_with_proxy_retry
)
from .gpt_client import gpt_response_message
from .document_builder import update_bd_and_create_document
__all__ = [
'download_proxies',
'get_shuffled_proxies',
'fetch_with_proxy',
'fetch_with_proxy_retry',
'gpt_response_message',
'update_bd_and_create_document'
]

78
services/document_builder.py Normal file
View File

@@ -0,0 +1,78 @@
"""
Document Builder - создание JSON и DOCX файлов
"""
import json
import os
from docx import Document
from config import DOCUMENTS_DIR
from utils import logger
import parser_bd as pbd
def update_bd_and_create_document(
response_text: str,
article_date: str,
url: str,
parsed_at: str,
original_text: str,
other: str
) -> None:
"""
Обрабатывает ответ от GPT, сохраняет в БД и создаёт DOCX документ
"""
clean_response = ''
if not response_text:
print(f"Пустой ответ от GPT для URL: {url}")
logger.info(f"Пустой ответ от GPT для URL: {url}")
return
try:
clean_response = response_text.strip().replace('```json', '').replace('```', '').strip()
data = json.loads(clean_response)
if data['category']:
data['article_date'] = article_date
data['url'] = url
data['parsed_at'] = parsed_at
data['original_text'] = original_text
data['status'] = False
data['viewed'] = False
data['other'] = other
# Сохранение в БД через pbd
parsed_data = pbd.ParsedData(**data)
pbd.save_parsed_data_to_db(parsed_data)
print("Данные успешно сохранены в БД")
# Создание DOCX документа
path_day = article_date.split()[0]
documents_path = os.path.join(DOCUMENTS_DIR, path_day)
if not os.path.exists(documents_path):
os.makedirs(documents_path)
print(f"Создана папка: {documents_path}")
doc = Document()
doc.add_heading('Ссылка на статью', level=1)
doc.add_paragraph(other)
doc.add_heading('Дата и время', level=1)
doc.add_paragraph(article_date)
doc.add_heading('Обнаруженные тематики текста', level=1)
doc.add_paragraph(data["category"])
doc.add_heading('Заголовок', level=1)
doc.add_paragraph(data["title"])
doc.add_heading('Краткий пересказ', level=1)
doc.add_paragraph(data["short_text"])
doc.add_heading('Переведенный текст статьи в газете', level=1)
doc.add_paragraph(data["translation_text"])
doc.add_heading('Оригинальный текст', level=1)
doc.add_paragraph(original_text)
doc_name = f"{data['title']}.docx"
doc_path = os.path.join(documents_path, doc_name)
doc.save(doc_path)
print(f"Сохранен документ: {doc_path}")
except Exception as ex:
print(f"Ошибка при обработке ответа GPT: {ex}")
logger.info(f"Ошибка при обработке ответа GPT: {ex}")

48
services/gpt_client.py Normal file
View File

@@ -0,0 +1,48 @@
"""
GPT клиент - отправка запросов к нейросети
"""
import time
import requests
from config import GPT_SERVER_URL, GPT_MAX_RETRIES, GPT_TIMEOUT
from utils import logger
import work_parser as wp
def gpt_response_message(content: str, name_promt: str) -> str:
"""
Отправляет текст на обработку GPT серверу
Возвращает ответ или пустую строку при ошибке
"""
contentGPT = wp.get_promt(name_promt).replace('{content}', content)
url = GPT_SERVER_URL
params = {'text': contentGPT}
max_retries = GPT_MAX_RETRIES
retries = 0
while retries < max_retries:
try:
response = requests.get(url, params=params, timeout=GPT_TIMEOUT)
return response.text
except requests.exceptions.ConnectTimeout as e:
print(f"Ошибка подключения (timeout): {e}")
logger.warning(f"gpt_response_message timeout:")
retries += 1
if retries < max_retries:
time.sleep(2 ** (retries - 1))
except requests.exceptions.ConnectionError as e:
print(f"Ошибка соединения: {e}")
logger.warning(f"gpt_response_message connection error: ")
retries += 1
if retries < max_retries:
time.sleep(2 ** (retries - 1))
except Exception as ex:
print(f"Ошибка при запросе к GPT: {ex}")
logger.error(f"gpt_response_message: ")
retries += 1
if retries < max_retries:
time.sleep(2 ** (retries - 1))
logger.info(f"Превышен лимит запросов {max_retries}")
return ""

75
services/proxy_manager.py Normal file
View File

@@ -0,0 +1,75 @@
"""
Менеджер прокси - управление загрузкой и использованием прокси
"""
import random
import requests
from config import PROXIES_URL
def download_proxies(url: str = PROXIES_URL) -> list[str]:
"""
Загружает список прокси из удаленного источника
"""
response = requests.get(url)
if response.status_code == 200:
proxies = response.text.splitlines()
return proxies
else:
return []
def get_shuffled_proxies(proxies_list: list[str]) -> list[str]:
"""
Перемешивает список прокси для случайного начала
"""
shuffled = proxies_list.copy()
random.shuffle(shuffled)
return shuffled
def fetch_with_proxy(url: str, proxy: str, verify: bool = True, timeout: int = 10) -> str | None:
"""
Выполняет запрос к URL через прокси
Возвращает текст ответа или None при ошибке
"""
proxies = {
'http': f'http://{proxy}',
'https': f'http://{proxy}',
}
try:
response = requests.get(url, proxies=proxies, timeout=timeout, verify=verify)
response.encoding = 'utf-8'
if response.status_code == 200:
# Проверяем содержимое - если это ошибка от прокси
if '"message":"Request failed' in response.text or '403' in response.text[:500]:
print(f"Proxy {proxy} - Site returned 403 (inside response)")
return None
print(f"Proxy {proxy} - SUCCESS")
return response.text
elif response.status_code == 403:
print(f"Proxy {proxy} - 403 Forbidden")
return None # Прокси работает, но сайт блокирует
else:
print(f"Proxy {proxy} - Status {response.status_code}")
return None
except:
return None
def fetch_with_proxy_retry(url: str, timeout: int = 10, verify: bool = True) -> str:
"""
Выполняет запрос с перебором прокси до успешного
Возвращает пустую строку если все прокси не сработали
"""
proxies_list = download_proxies(PROXIES_URL)
proxies_list = get_shuffled_proxies(proxies_list)
response = ""
for proxy in proxies_list:
response = fetch_with_proxy(url, proxy=proxy, timeout=timeout, verify=verify)
if response:
break
else:
response = ""
return response

12
utils/__init__.py Normal file
View File

@@ -0,0 +1,12 @@
"""
Утилиты приложения
"""
from .logger import setup_logger, logger
from .helpers import create_folder, get_current_date_parts
__all__ = [
'setup_logger',
'logger',
'create_folder',
'get_current_date_parts'
]

25
utils/helpers.py Normal file
View File

@@ -0,0 +1,25 @@
"""
Общие вспомогательные функции
"""
from datetime import datetime as dt
def create_folder(num: int) -> str:
"""
Форматирует номер дня/месяца для имени папки (добавляет ведущий ноль)
"""
if int(num) // 10 == 0:
return f"0{num}"
else:
return str(num)
def get_current_date_parts() -> tuple[str, str, str]:
"""
Возвращает текущую дату в формате (год, месяц, день) с форматированием
"""
datetime_now = dt.now()
current_day = create_folder(datetime_now.day)
current_month = create_folder(datetime_now.month)
current_year = f"{datetime_now.year}"
return current_year, current_month, current_day

22
utils/logger.py Normal file
View File

@@ -0,0 +1,22 @@
"""
Настройка логгера
"""
import logging
from config import LOG_FILE
def setup_logger(name: str = __name__) -> logging.Logger:
"""
Настройка и возврат логгера
"""
logging.basicConfig(
filename=LOG_FILE,
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(name)
return logger
# Глобальный логгер
logger = setup_logger()