"""Module to define utility function""" import os import re import json import uuid import time import glob import urllib.request from urllib.parse import urlparse from datetime import datetime, timedelta from decimal import Decimal import pandas as pd import requests import boto3 from lxml import etree from dotenv import load_dotenv from googletrans import Translator from transformers import pipeline from PyPDF2 import PdfReader from langdetect import detect load_dotenv() AWS_ACCESS_KEY_ID = os.environ['AWS_ACCESS_KEY_ID'] AWS_SECRET_ACCESS_KEY = os.environ['AWS_SECRET_ACCESS_KEY'] # AWS_ACCESS_KEY_ID = "AKIAQFXZMGHQYXKWUDWR" # AWS_SECRET_ACCESS_KEY = "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" analyzer = pipeline("sentiment-analysis", model="ProsusAI/finbert") translator = Translator() with open('xpath.json', 'r', encoding='UTF-8') as f: xpath_dict = json.load(f) with open('patterns.json', 'r', encoding='UTF-8') as f: patterns = json.load(f) def get_client_connection(): """ Returns a client connection to DynamoDB. :return: DynamoDB client connection """ dynamodb = boto3.client( service_name='dynamodb', region_name='us-east-1', aws_access_key_id=AWS_ACCESS_KEY_ID, aws_secret_access_key=AWS_SECRET_ACCESS_KEY ) return dynamodb def update_reference(report): """ Updates the reference in the 'reference_china' table in DynamoDB. Args: report (dict): A dictionary containing the report details. Returns: None """ dynamodb = get_client_connection() response = dynamodb.update_item( TableName="reference_china", Key={ 'id': {'S': str(report['refID'])}, 'sourceID': {'S': str(report['sourceID'])} }, UpdateExpression='SET link = :link, referenceID = :referenceID, LastModifiedDate = :LastModifiedDate', ExpressionAttributeValues={ ':link': {'S': report['link']}, ':referenceID': {'S': report['referenceID']}, ':LastModifiedDate': {'S': datetime.now().strftime("%Y-%m-%dT%H:%M:%S")}, } ) print(response) def download_files_from_s3(folder): """ Downloads Parquet files from an S3 bucket and returns a concatenated DataFrame. Args: folder (str): The folder in the S3 bucket to download files from. Returns: pandas.DataFrame: A concatenated DataFrame containing the data from the downloaded Parquet files. """ if not os.path.exists(folder): os.makedirs(folder) client = boto3.client( 's3', aws_access_key_id=AWS_ACCESS_KEY_ID, aws_secret_access_key=AWS_SECRET_ACCESS_KEY, ) response = client.list_objects_v2(Bucket='china-securities-report', Prefix=f"{folder}/") for obj in response['Contents']: key = obj['Key'] if key.endswith('.parquet'): client.download_file('china-securities-report', key, key) file_paths = glob.glob(os.path.join(folder, '*.parquet')) return pd.concat([pd.read_parquet(file_path) for file_path in file_paths], ignore_index=True) def extract_from_pdf_by_pattern(url, pattern): """ Extracts text from a PDF file based on a given pattern. Args: url (str): The URL of the PDF file to extract text from. pattern (dict): A dictionary containing the pattern to match and the pages to extract text from. Returns: str: The extracted text from the PDF file. Raises: Exception: If there is an error while retrieving or processing the PDF file. """ # Send a GET request to the URL and retrieve the PDF content try: response = requests.get(url, timeout=60) pdf_content = response.content # Save the PDF content to a local file with open("downloaded_file.pdf", "wb") as file: file.write(pdf_content) # Open the downloaded PDF file and extract the text with open("downloaded_file.pdf", "rb") as file: pdf_reader = PdfReader(file) extracted_text = "" if 'pages' in pattern: pages = pattern['pages'] else: pages = len(pdf_reader.pages) for page in pages: text = pdf_reader.pages[page].extract_text() if 'keyword' in pattern and pattern['keyword'] in text: text = text.split(pattern['keyword'], 1)[1].strip() else: text = text.strip() extracted_text += text except: extracted_text = '' return extracted_text.replace('?\n', '?-\n').replace('！\n', '！-\n').replace('。\n', '。-\n').replace('\n',' ').replace('?-','?\n').replace('！-','！\n').replace('。-','。\n') def get_reference_by_regex(pattern, text): """ Finds all occurrences of a given regex pattern in the provided text. Args: pattern (str): The regex pattern to search for. text (str): The text to search within. Returns: list: A list of all matches found in the text. """ return re.findall(pattern, text) def isnot_substring(list_a, string_to_check): """ Check if any string in the given list is a substring of the string_to_check. Args: list_a (list): A list of strings to check. string_to_check (str): The string to check for substrings. Returns: bool: True if none of the strings in list_a are substrings of string_to_check, False otherwise. """ for s in list_a: if s in string_to_check: return False return True def extract_reference(row): """ Extracts reference information from a given row. Args: row (dict): A dictionary representing a row of data. Returns: None """ try: pattern = next((elem for elem in patterns if elem['site'] == row['site']), None) extracted_text = extract_from_pdf_by_pattern(row['attachment'],pattern) reference_titles = re.findall(pattern['article_regex'], extracted_text) reference_dates = re.findall(pattern['date_regex'], extracted_text) reference_titles = [s.replace(' ', '') for s in reference_titles] reference_dates = [s.replace(' ', '') for s in reference_dates] print(reference_dates, reference_titles) if 'remove' in pattern: for remove_string in pattern['remove']: reference_titles = [s.replace(remove_string, '') for s in reference_titles] if len(reference_dates) > 0: for title, date in zip(reference_titles, reference_dates): try: date = datetime.strptime(date, pattern['date_format']) except: date = datetime(2006, 1, 1) dates = [] if 'date_range' in pattern: for i in range(pattern['date_range'] + 1): dates.append((date + timedelta(days=i)).strftime('%Y-%m-%d')) dates.append((date - timedelta(days=i)).strftime('%Y-%m-%d')) dates.append(date.strftime('%Y-%m-%d')) date = date.strftime('%Y-%m-%d') if 'split' in pattern: for split_item in pattern['split']: if 'exceptional_string' in split_item: if split_item['string'] in title and isnot_substring(split_item['exceptional_string'], title): title = re.split(split_item['string'], title)[split_item['index']] else: if split_item['string'] in title: title = title.split(split_item['string'])[split_item['index']] if len(data[(data['titleCN'].str.contains(title)) & (data['site'] == row['site']) & (data['publishdate'].isin(dates))]) == 0: print("------------ = 0 ------------") print(date, repr(title)) elif len(data[(data['titleCN'].str.contains(title)) & (data['site'] == row['site']) & (data['publishdate'].isin(dates))]) > 1: print("------------ > 1 ------------") print(date, repr(title)) else: print("------------ = 1 ------------") reference_df = data[(data['titleCN'].str.contains(title)) & (data['site'] == row['site']) & (data['publishdate'].isin(dates))] row['referenceID'] = reference_df.iloc[0]['id'] row['link'] = reference_df.iloc[0]['link'] row['sourceID'] = row['id'] row['refID'] = uuid.uuid5(uuid.NAMESPACE_OID, str(row['sourceID'])+str(row['referenceID'])) print(date, repr(title), row['sourceID'],row['referenceID']) update_reference(row) else: for title in reference_titles: if 'split' in pattern: for split_item in pattern['split']: if 'exceptional_string' in split_item: if split_item['string'] in title and isnot_substring(split_item['exceptional_string'], title): title = re.split(split_item['string'], title)[split_item['index']] else: if split_item['string'] in title: title = title.split(split_item['string'])[split_item['index']] if len(data[(data['titleCN'].str.contains(title)) & (data['site'] == row['site'])]) == 0: print("------------ = 0 ------------") print(repr(title)) elif len(data[(data['titleCN'].str.contains(title)) & (data['site'] == row['site'])]) > 1: print("------------ > 1 ------------") print(repr(title)) else: print("------------ = 1 ------------") reference_df = data[(data['titleCN'].str.contains(title)) & (data['site'] == row['site'])] row['referenceID'] = reference_df.iloc[0]['id'] row['link'] = reference_df.iloc[0]['link'] row['sourceID'] = row['id'] row['refID'] = uuid.uuid5(uuid.NAMESPACE_OID, str(row['sourceID'])+str(row['referenceID'])) print(repr(title), row['sourceID'],row['referenceID']) update_reference(row) except Exception as error: print(error) def translate(text): """ Translates the given text to English. Args: text (str): The text to be translated. Returns: str: The translated text in English. """ return translator.translate(text, dest='en').text def datemodifier(date_string, date_format): """Date Modifier Function This function takes a date string and a date format as input and modifies the date string according to the specified format. It returns the modified date string in the format 'YYYY-MM-DD'. Args: date_string (str): The date string to be modified. date_format (str): The format of the date string. Returns: str: The modified date string in the format 'YYYY-MM-DD'. False: If an error occurs during the modification process. """ try: to_date = time.strptime(date_string,date_format) return time.strftime("%Y-%m-%d",to_date) except: return False def fetch_url(url): """ Fetches the content of a given URL. Args: url (str): The URL to fetch. Returns: str or None: The content of the URL if the request is successful (status code 200), otherwise None. Raises: requests.exceptions.RequestException: If there is an error while making the request or if the response status code is not 200. """ try: response = requests.get(url, timeout=60) if response.status_code == 200: return response.text else: return None except requests.exceptions.RequestException or requests.exceptions.ReadTimeout as e: print(f"An error occurred: {e}") # Optional: handle or log the error in some way return None def translist(infolist): """ Filter and transform a list of strings. Args: infolist (list): The input list of strings. Returns: list: The filtered and transformed list of strings. """ out = list(filter(lambda s: s and (isinstance(s, str) or len(s.strip()) > 0), [i.strip() for i in infolist])) return out def encode(content): """ Encodes the given content into a single string. Args: content (list): A list of elements to be encoded. Each element can be either a string or an `etree._Element` object. Returns: str: The encoded content as a single string. """ text = '' for element in content: if isinstance(element, etree._Element): subelement = etree.tostring(element).decode() subpage = etree.HTML(subelement) tree = subpage.xpath('//text()') line = ''.join(translist(tree)).\ replace('\n','').replace('\t','').replace('\r','').replace(' ','').strip() else: line = element text += line return text def encode_content(content): """ Encodes the content by removing unnecessary characters and extracting a summary. Args: content (list): A list of elements representing the content. Returns: tuple: A tuple containing the encoded text and the summary. """ text = '' for element in content: if isinstance(element, etree._Element): subelement = etree.tostring(element).decode() subpage = etree.HTML(subelement) tree = subpage.xpath('//text()') line = ''.join(translist(tree)).\ replace('\n','').replace('\t','').replace('\r','').replace(' ','').strip() else: line = element if line != '': line = line + '\n' text += line index = text.find('打印本页') if index != -1: text = text[:index] try: summary = '\n'.join(text.split('\n')[:2]) except: summary = text return text, summary def extract_from_pdf(url): """ Extracts text from a PDF file given its URL. Args: url (str): The URL of the PDF file. Returns: tuple: A tuple containing the extracted text and a summary of the text. Raises: Exception: If there is an error during the extraction process. """ # Send a GET request to the URL and retrieve the PDF content response = requests.get(url, timeout=60) pdf_content = response.content # Save the PDF content to a local file with open("downloaded_file.pdf", "wb") as file: file.write(pdf_content) # Open the downloaded PDF file and extract the text with open("downloaded_file.pdf", "rb") as file: pdf_reader = PdfReader(file) num_pages = len(pdf_reader.pages) extracted_text = "" for page in range(num_pages): text = pdf_reader.pages[page].extract_text() if text and text[0].isdigit(): text = text[1:] # first_newline_index = text.find('。\n') # text = text[:first_newline_index+1].replace('\n', '') + text[first_newline_index+1:] text = text.replace('?\n', '?-\n').replace('！\n', '！-\n').replace('。\n', '。-\n').replace('\n','').replace('?-','?\n').replace('！-','！\n').replace('。-','。\n') if text != '': extracted_text += text try: summary = '\n'.join(extracted_text.split('\n')[:2]) except: summary = text return extracted_text, summary def get_db_connection(): """Get dynamoDB connection. Returns: boto3.resource: The DynamoDB resource object representing the connection. """ dynamodb = boto3.resource( service_name='dynamodb', region_name='us-east-1', aws_access_key_id=AWS_ACCESS_KEY_ID, aws_secret_access_key=AWS_SECRET_ACCESS_KEY ) return dynamodb def sentiment_computation(content): """ Compute the sentiment score and label for the given content. Parameters: content (str): The content for which sentiment needs to be computed. Returns: tuple: A tuple containing the sentiment score and label. The sentiment score is a float representing the overall sentiment score of the content. The sentiment label is a string representing the sentiment label ('+', '-', or '0'). """ label_dict = { "positive": "+", "negative": "-", "neutral": "0", } sentiment_score = 0 maximum_value = 0 raw_sentiment = analyzer(content[:511], top_k=None) sentiment_label = None for sentiment_dict in raw_sentiment: value = sentiment_dict["score"] if value > maximum_value: sentiment_label = sentiment_dict["label"] maximum_value = value if sentiment_dict["label"] == "positive": sentiment_score = sentiment_score + value if sentiment_dict["label"] == "negative": sentiment_score = sentiment_score - value else: sentiment_score = sentiment_score + 0 sentiment_score = sentiment_score * 100 return sentiment_score, label_dict[sentiment_label] def crawl(url, article): """ Crawls the given URL and extracts relevant information from the webpage. Args: url (str): The URL of the webpage to crawl. article (dict): A dictionary to store the extracted information. Returns: None: If the length of the extracted content is less than 10 characters. Raises: None """ domain = '.'.join(urlparse(url).netloc.split('.')[1:]) req = urllib.request.urlopen(url) text = req.read() html_text = text.decode("utf-8") page = etree.HTML(html_text) contentCN, summary = encode_content(page.xpath(xpath_dict[domain]['content'])) article['originSite'] = xpath_dict[domain]['siteCN'] article['site'] = xpath_dict[domain]['site'] article['titleCN'] = encode(page.xpath(xpath_dict[domain]['title'])) article['title'] = translate(article['titleCN']) if 'author' in xpath_dict[domain]: article['author'] = translate(encode(page.xpath(xpath_dict[domain]['author']))) else: article['author'] = "" article['contentCN'] = repr(contentCN)[1:-1].strip() if len(article['contentCN']) < 10: return None CONTENT_ENG = '' for element in contentCN.split("\n"): CONTENT_ENG += translate(element) + '\n' if detect(CONTENT_ENG) != 'en': for element in contentCN.split("。"): CONTENT_ENG += translate(element) + '. ' article['content'] = repr(CONTENT_ENG)[1:-1].strip() if 'subtitle' in xpath_dict[domain]: article['subtitle'] = translate(encode(page.xpath(xpath_dict[domain]['subtitle']))) else: article['subtitle'] = translate(summary) article['publishDate'] = datemodifier(encode(page.xpath(xpath_dict[domain]['publishdate'])), xpath_dict[domain]['datetime_format']) article['link'] = url article['attachment'] = "" article['sentimentScore'], article['sentimentLabel'] = sentiment_computation(CONTENT_ENG.replace("\n","")) article['id'] = uuid.uuid5(uuid.NAMESPACE_OID, article['titleCN']+article['publishDate']) print(article['id'], article['site'] ) update_content(article) def upsert_content(report): """ Upserts the content of a report into the 'article_china' table in DynamoDB. Args: report (dict): A dictionary containing the report data. Returns: dict: The response from the DynamoDB put_item operation. """ dynamodb = get_db_connection() table = dynamodb.Table('article_china') # Define the item data item = { 'id': str(report['id']), 'site': report['site'], 'title': report['title'], 'titleCN': report['titleCN'], 'contentCN': report['contentCN'], 'category': report['category'], 'author': report['author'], 'content': report['content'], 'subtitle': report['subtitle'], 'publishDate': report['publishDate'], 'link': report['link'], 'attachment': report['attachment'], # 'authorID': str(report['authorid']), # 'entityList': report['entitylist'], 'sentimentScore': Decimal(str(report['sentimentScore'])).quantize(Decimal('0.01')), 'sentimentLabel': report['sentimentLabel'], 'LastModifiedDate': datetime.now().strftime("%Y-%m-%dT%H:%M:%S") } response = table.put_item(Item=item) print(response) def delete_records(item): """ Deletes a record from the 'article_test' table in DynamoDB. Args: item (dict): The item to be deleted, containing 'id' and 'site' keys. Returns: None """ dynamodb_client = get_client_connection() dynamodb_client.delete_item( TableName="article_test", Key={ 'id': {'S': item['id']}, 'site': {'S': item['site']} } ) def update_content(report): """ Updates the content of an article in the 'article_china' table in DynamoDB. Args: report (dict): A dictionary containing the report data. Returns: None """ dynamodb = get_client_connection() response = dynamodb.update_item( TableName="article_china", Key={ 'id': {'S': str(report['id'])}, 'site': {'S': report['site']} }, UpdateExpression='SET title = :title, titleCN = :titleCN, contentCN = :contentCN, category = :category, author = :author, content = :content, subtitle = :subtitle, publishDate = :publishDate, link = :link, attachment = :attachment, sentimentScore = :sentimentScore, sentimentLabel = :sentimentLabel, LastModifiedDate = :LastModifiedDate', ExpressionAttributeValues={ ':title': {'S': report['title']}, ':titleCN': {'S': report['titleCN']}, ':contentCN': {'S': report['contentCN']}, ':category': {'S': report['category']}, ':author': {'S': report['author']}, ':content': {'S': report['content']}, ':subtitle': {'S': report['subtitle']}, ':publishDate': {'S': report['publishDate']}, ':link': {'S': report['link']}, ':attachment': {'S': report['attachment']}, ':LastModifiedDate': {'S': datetime.now().strftime("%Y-%m-%dT%H:%M:%S")}, ':sentimentScore': {'N': str(Decimal(str(report['sentimentScore'])).quantize(Decimal('0.01')))}, ':sentimentLabel': {'S': report['sentimentLabel']} } ) print(response) def update_content_sentiment(report): """ Updates the sentiment score and label of an article in the 'article_test' DynamoDB table. Args: report (dict): A dictionary containing the report information. Returns: None """ dynamodb = get_client_connection() response = dynamodb.update_item( TableName="article_test", Key={ 'id': {'S': report['id']}, 'site': {'S': report['site']} }, UpdateExpression='SET sentimentScore = :sentimentScore, sentimentLabel = :sentimentLabel', ExpressionAttributeValues={ ':sentimentScore': {'N': str(Decimal(str(report['sentimentscore'])).quantize(Decimal('0.01')))}, ':sentimentLabel': {'S': report['sentimentlabel']} } ) print(response) data = download_files_from_s3('data')