import lxml.etree import requests import aiohttp import asyncio import ListOfSities import configparser from time import sleep import os import sys import getopt import platform import tkinter.filedialog from random import randint from lxml import etree import lxml # debugging part # import ssl # from aiologger import Logger # from icecream import ic # import llm provider # from freeGPT import Client from g4f.client import Client from asyncio import run import time from docxtpl import DocxTemplate from bs4 import BeautifulSoup # Logger.basicConfig(level=logger.INFO, filename=f"{time.localtime()}") # Some debugging options... class bcolors: HEADER = '\033[95m' OKBLUE = '\033[94m' OKCYAN = '\033[96m' OKGREEN = '\033[92m' WARNING = '\033[93m' FAIL = '\033[91m' ENDC = '\033[0m' BOLD = '\033[1m' UNDERLINE = '\033[4m' async def fetch_page(session, url, headers): # make GET request using session async with session.get(url, headers=headers) as response: # return HTML content return await response.text(), url async def get_news_text(urls, headers, limit : int) -> dict: newslist = {} # intCountIter = len(urls) // 5 # if len(urls) / 5 != intCountIter: # intCountIter += 1 # throttle aiohttp bruh connector = aiohttp.TCPConnector(limit=limit) async with aiohttp.ClientSession(headers=headers, connector=connector) as session: # Initialize tasks list tasks = [] # loop through URLs and append tasks for url in urls: # await print(f"url: {url}, /n intCountIter: {intCountIter}") tasks.append(fetch_page(session, url, headers=headers)) # group and Execute tasks concurrently htmls = await asyncio.gather(*tasks) for url, html in zip(urls, htmls): newslist.update({url : html}) return newslist def chDir() -> str: sep = '' if platform.system() != 'Linux': tempi = '' for i in time.localtime()[1:5]: tempi += str(i) try: os.makedirs(f"{tempi}-inform-pack") except: pass sep += f'\\{tempi}-inform-pack\\' else: sep = "/" return sep def preparing() -> list: print(f"{bcolors.OKBLUE}{bcolors.BOLD}Валек который делает информирование вовремя \n Билд: 2.0.6 от 06.07.24(дембельский){bcolors.ENDC}") print(f"{bcolors.OKCYAN}{bcolors.BOLD}govnocoded by valo\n Discord: valo#1234 | @valoker\n Telegram: @valoka{bcolors.ENDC}") # dateArr = '' # Remove 1st argument from the # list of command line arguments try: argumentList = sys.argv[1:] if len(argumentList) == 0: raise getopt.error("Аргументов нет") # Options options = "hmo:" # Long options long_options = ["Help", "Input="] # Parsing argument arguments, values = getopt.getopt(argumentList, options, long_options) # checking each argument for currentArgument, currentValue in arguments: if currentArgument in ("-h", "--help"): print("Nothing Helpful") elif currentArgument in ("-i", "--Input"): print(("В качестве аргумента было полученно: (% s)") % (currentValue)) dateArr = str(currentValue) # ???? # print(dateArr) # time.sleep(10) return dateArr.split(',') except getopt.error as err: # output error, and return with an error code print(str(err)) # ссылки типа: https://ria.ru/20240214/ # применять только архивы dateArr = input(f"{bcolors.BOLD}Введите дату (ГГГГММДД): {bcolors.ENDC}").split(',') return dateArr def getConf() -> list: config = configparser.ConfigParser() try: config.read("Config.ini", encoding='utf-8-sig') totalGuys = int(config["Main"]["totalGuys"]) filepath = config["Path"]["PathToFile"] noAI = config["Main"]["noAI"] limit = int(config["Main"]["limit"]) if config["Path"]["PathToFile"] == "None": filepath = tkinter.filedialog.askopenfilename() except Exception as e: print("err:", e) print(f"""{bcolors.FAIL}Отсутствует фаил конфигурации. \n Укажите путь до файла шаблона вручную..{bcolors.ENDC}""") filepath = tkinter.filedialog.askopenfilename() return totalGuys, config, filepath, noAI, limit # def askai(question): # g4f.debug.logging = False # g4f.check_version = False # return Client.create_completion("gpt3", question) def askai(question) -> str: #client.debug.logging = False #client.check_version = False client = Client() response = client.chat.completions.create( model="gpt-3.5-turbo", messages=[{"role": "user", "content": f"{question}"}], ) return str(response.choices[0].message.content) st_accept = "text/html" st_useragent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:123.0) Gecko/20100101 Firefox/123.0" headers = { "Accept": st_accept, "User-Agent": st_useragent} def getUpdateLink(url : str): '''Returns updated link of main page. Gets: latest link ''' # gets: https://crimea.ria.ru/20240501/rossiyskie-voyska...-1136972211.html # returns: https://ria.ru/services/20240501/more.html?id=1943375808&date=20240501T223909 dataId = '' dataPublished = '' # linkdata = [] # tags = [] page = requests.get(url=url, timeout=10) soup = BeautifulSoup(page.text, "lxml") # Bruh moment #1 tempDataId = url.split('-') dataId = tempDataId[len(dataId) - 1].split('.')[0] del tempDataId # cpp moment # Bruh moment #2 dataPublished = soup.find('div', itemprop='datePublished').get_text() dataPublished = dataPublished.replace('-', '').replace(':', '') # Construct final url temporalUrl = url.split('/') finalUrl = '/'.join(temporalUrl[0:3]) finalUrl = f"{finalUrl}/services/{temporalUrl[3]}/more.html?id={dataId}&date={dataPublished}'" return finalUrl def getUpdateUrls(url : str) -> dict: '''Get some news (links and newsheaders)''' updatedUrl = getUpdateLink(url) mainPage = requests.get(updatedUrl, timeout=10) # links = [] # newsheaders = [] NLDict = {} soup = BeautifulSoup(mainPage.text, "lxml") allNews = soup.findAll("a", class_="list-item__title color-font-hover-only") soupedAllNews = BeautifulSoup(str(allNews), "lxml") for data in soupedAllNews.findAll(href=True): # links.append(data['href']) # newsheaders.append(data.get_text()) NLDict.update({data['href'] : data.get_text()}) return NLDict # default values noAI = True limit = 7 def process(url, date, totalGuys, config, filepath, headers=headers, noAI=noAI, limit=limit): try: try: print(f"{bcolors.OKCYAN}Стучусь до РИА...{bcolors.ENDC}") mainPage = requests.get(url, timeout=10) print(f"{bcolors.OKGREEN}Страница получена{bcolors.ENDC}") except requests.exceptions.Timeout: print(f"{bcolors.FAIL}Превышено время ожидания{bcolors.ENDC}") exit(1) allNews = [] soup = BeautifulSoup(mainPage.text, "lxml") allNews = soup.findAll("a", itemprop='url') soupedAllNews = BeautifulSoup(str(allNews), features="lxml") links = [] #get links for data in soupedAllNews.findAll(href=True): string = data['href'] # remove url[:-1] links.append(string) links = links[4:] #get news headers newsheaders = [] for data in soup.findAll("a", class_="list-item__title color-font-hover-only"): newsheaders.append(data.get_text()) # ithink 40 is enough updatedNews = getUpdateUrls(links[len(links) - 1]) # get more news links = links + list(updatedNews.keys()) newsheaders = newsheaders + list(updatedNews.values()) # asking ai for cool news for guyNum in range(0, totalGuys): newsheadersstring = "$".join(newsheaders) print(f"{bcolors.OKCYAN}Сборка конспекта для: {bcolors.OKGREEN}{config[f"Utverzhdau_{guyNum}"]["subdivision"]}{bcolors.ENDC}") # думаю тут нужно использовать tuple if len(newsheaders) != len(links): asd = '' zxc = [str(i) for i in range(1, len(links) + 1)] for i in zxc: if i is not zxc[len(zxc) - 1]: asd = asd + i + ', ' else: asd = asd + i print(f"{bcolors.OKCYAN}Количество заголовков/ссылок:", f'{bcolors.OKGREEN}{len(newsheaders)} {bcolors.FAIL + f"\nПохоже, что ссылок больше... Cкорее всего разметка сайта снова сменилась. \nЯзыковая модель не будет помогать в этом запуске. " if len(newsheaders) != len(links) else bcolors.OKGREEN}Количество ссылок: {len(links)}{bcolors.ENDC}') else: count = 0 while True: try: if noAI == False: asd = askai(f"""Есть несколько новостных заголовков, но нужно самые интересные из них разместить в начале, ответ представь только в виде чисел через запятую и пробелом после запятой, ответ не должен содержать буквы, только запятые и цифры. После каждой цифры должен применяться символ ','. Каждый заголовок отделен $, также помести в конец новости которые содержат какие-либо упомнинания о войнах или содержат слова по типу: СВО, ВСУ, спецоперация, ДНР, ЛНР, Украинские военные, Украинские боевики, Украинские и им подобные, После двоеточия перечисленны новостные заголовки, вот они: {newsheadersstring}""").strip() else: asd = '' zxc = [str(i) for i in range(1, len(links) + 1)] for i in zxc: if i is not zxc[len(zxc) - 1]: asd = asd + i + ', ' else: asd = asd + i #asd = "1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19" print(f"{bcolors.OKCYAN}{'Вывод языковой модели:' if noAI == False else 'Получены новости:'} {bcolors.OKGREEN}{asd}{bcolors.ENDC}") if int(asd[0]) % 1 == 0 and int(asd[len(asd)-1]) % 1 == 0: break except ValueError: count += 1 print(f"{bcolors.FAIL}Неинтерпритируеымй ответ от железяки, пробую еще раз...{bcolors.ENDC}") if count > 3: print(f"Слишком много вызовов к модели, повторю попытку через 15 сек...") sleep(5) if count > 5: # можешь ты хоть раз себя не повторять? asd = '' zxc = [str(i) for i in range(1, len(links))] for i in zxc: if i is not zxc[len(zxc) - 1]: asd = asd + i + ', ' else: asd = asd + i break tempNewsLinks = links links = [] goodNews = asd.split(", ") #WTF? but its ok for i in range(0, len(goodNews)): if int(goodNews[i]) < len(tempNewsLinks): links.append(tempNewsLinks[int(goodNews[i]) - 1]) else: print("Модель вернула больше новостей, чем есть") # for i in goodNews: # links.append(int(i) - 1) # for i in links: # temptempnews.append(tempNewsLinks[i]) # links = temptempnews print(f"{bcolors.OKCYAN}Получено новостей: {bcolors.OKGREEN}{len(links)}{bcolors.ENDC}") newsCity = [] textNews = '' foreignNews = '' countWords = 0 tempNews = [] #get news newsAbroad = False anotherStupidFlag = False newslist = [] newslist = (asyncio.run(get_news_text(links, headers=headers, limit=limit))) sortedNews = [] ''' написать сортировщик, который берет юрл из links и ставит такой же порядок ссылок как и в newslist''' # stupid way # лишняя переменная for link in links: sortedNews.append(newslist.get(link)) #debug # with open('wet.txt', 'w') as f: # for i, o in sortedNews: # try: # print(i, o, file=f) # except UnicodeEncodeError: # pass for page, url in sortedNews: soup = BeautifulSoup(str(page), "html.parser") try: text = soup.find('div', itemprop='articleBody').get_text() #print(link, text) except: continue newsCity.append(ListOfSities.RusNews(text.split(",", 1)[0])) if ListOfSities.RusNews(text.split(",", 1)[0]) == False: tempNews.append(text + '') if newsAbroad == True: foreignNews += text + '' anotherStupidFlag = True countWords += len(textNews.split(" ")) if countWords > 890: newsAbroad = False break else: if countWords <= 790 and newsAbroad == False: textNews += text + '' countWords += len(textNews.split(" ")) else: if newsAbroad == False: newsAbroad = True if (foreignNews == '' and newsAbroad == True) or countWords < 750: print(f'{bcolors.WARNING}В конечный фаил будет добавлена дополнительная новость, поскольку УВ2 пуст (количество слов, наличие УВ2){bcolors.ENDC}', countWords, anotherStupidFlag) for news in tempNews: if countWords < 890 or anotherStupidFlag == False: anotherStupidFlag = True foreignNews += news + '' #count lines totalLines = int(len(textNews) // 30) manyIndent = '' * totalLines #debug #print(newsCity) #print(foreignNews) #print(textNews) #print(totalLines) #print(countWords) #docx processing try: doc = DocxTemplate(filepath) except Exception as e: print("err", e) exit(1) context = { 'news_text' : f"{textNews}", 'empty_line' : f"{manyIndent}", 'service_position' : f'{config[f"Utverzhdau_{guyNum}"]["commander"]}', 'rank' : f'{config[f"Utverzhdau_{guyNum}"]["rank"]}', 'rank' : f'{config[f"Utverzhdau_{guyNum}"]["rank"]}', # valek is sus 'full_name' : f'{config[f"Utverzhdau_{guyNum}"]["name"].replace('"', '') if randint(83, 100) > 83 else 'В.А. Дорофеев'}', 'day' : f'{date[6:8]}', 'month' : f'{date[4:6]}', 'year' : f'{date[:4]}', #some hardcode 'subdivision' : f'{config[f"Utverzhdau_{guyNum}"] ["subdivision"].replace(' ', ' ' * int(config[f"Utverzhdau_{guyNum}"]["spaces"]))}', 'plan_subdivision' : f'{config[f"Utverzhdau_{guyNum}"]['subdivision']}', 'plan_squad' : f'{config["Plan"]["squad"] if config[f"Utverzhdau_{guyNum}"]['subdivision'] != "телекоммуникационного узла" else " "}', 'foreign_text' : f"{foreignNews}"} doc.render(context) sep = chDir() fullPath = f"{os.getcwd()}{sep}gen-{config[f"Utverzhdau_{guyNum}"]['subdivision']}-{date}.docx" print(fullPath) doc.save(fullPath) os.system('"' + fullPath + '"') except BaseException as err: print(err) input("enter any key to exit") exit(1)