PyPI 官网下载 | workplace_extractor-0.7.16.5.tar.gz

from workplace_extractor.Extractors.GroupExtractor import GroupExtractor from workplace_extractor.Extractors.PersonExtractor import PersonExtractor from workplace_extractor.Nodes.NodeCollection import PostCollection from workplace_extractor.Nodes.Author import Bot from workplace_extractor.Nodes.Post import Summary, Post from workplace_extractor.Nodes.Interaction import View, Reaction, Comment from copy import copy import logging import pandas as pd import networkx as nx import matplotlib.pyplot as plt import matplotlib from random import randint class PostExtractor: def __init__(self, extractor, since, until): self.extractor = extractor self.since = since self.until = until self.nodes = PostCollection() self.filter_ids = [] async def extract(self): logging.info('Loading group feeds') group_extractor = GroupExtractor(self.extractor) await group_extractor.extract() self.nodes.extend(group_extractor.nodes) logging.info('Group feeds loaded') logging.info('Loading person feeds') people_extractor = PersonExtractor(self.extractor) await people_extractor.extract() self.nodes.extend(people_extractor.nodes) logging.info('Person feeds loaded') # fields that should be extracted from posts using the GRAPH API fields = 'id,from,type,created_time,status_type,object_id,link,message,story' logging.info(f'Extracting posts from {len(group_extractor.nodes.nodes)} groups.') http_calls = [] for node in group_extractor.nodes.nodes: http_calls.append({'url': self.extractor.config.get('URL', 'GRAPH') + f'/{node.node_id}/feed?limit=100' f'&fields={fields}' f'&since={self.since}' f'&until={self.until}', 'call': self.call, 'node': node, 'recursion': 1}) await self.extractor.fetch(http_calls) logging.info(f'Extracting posts from {len(people_extractor.nodes.nodes)} people.') http_calls = [] for node in people_extractor.nodes.nodes: http_calls.append({'url': self.extractor.config.get('URL', 'GRAPH') + f'/{node.node_id}/feed?limit=100' f'&fields={fields}' f'&since={self.since}' f'&until={self.until}', 'call': self.call, 'node': node, 'recursion': 1}) await self.extractor.fetch(http_calls) count = 0 for node in self.nodes.nodes: count += len(node.feed.nodes) # Add view, reactions, comments and authors await self.fetch_posts_info() logging.info('Post extraction ended') async def fetch_posts_info(self): http_calls = [] for node in self.nodes.nodes: for post in node.feed.nodes: http_calls.append({ 'url': self.extractor.config.get('URL', 'GRAPH') + f'/{post.node_id}?fields=' 'reactions.limit(100).summary(1),' 'seen.limit(100).summary(1),' 'comments.limit(100).summary(1)' '{' 'created_time,' 'from,' 'reactions.limit(100).summary(1),' 'comments.limit(100).summary(1)' '{' 'created_time,' 'from,' 'reactions.limit(100).summary(1)' '}' '}', 'call': self.call_info if self.extractor.export == 'INTERACTIONS' else self.call_count, 'post': post, 'recursion': 1}) author = next((node for node in self.nodes.nodes if node.node_id == post.author_id), None) # including Bots if author is not None: author = copy(author) author.feed = None post.author = author else: http_calls.append({'url': self.extractor.config.get('URL', 'GRAPH') + f'/{post.author_id}', 'call': self.call_bot, 'post': post}) await self.extractor.fetch(http_calls) async def call(self, url, session, **kwargs): data = await self.extractor.fetch_url(url, session, 'GRAPH', **kwargs) if 'data' in data and data['data']: collection = PostCollection([Post(post) for post in data['data']]) # filter posts already extracted. Applies to posts in both group and author feed and for # GRAPH bug with infinite loop. collection.set_partial_id() collection.drop_duplicates(self.filter_ids) # if hashtags not empty, filter result by hashtag if self.extractor.hashtags: collection.filter_hashtags(self.extractor.hashtags) kwargs.get('node').feed.extend(collection) self.add_to_filter(collection.nodes) next_page = data.get('paging', {}).get('next') if next_page is not None: kwargs['recursion'] += 1 await self.call(next_page, session, **kwargs) async def call_info(self, url, session, **kwargs): data = await self.extractor.fetch_url(url, session, 'GRAPH', **kwargs) await self.call_info_views(data.get('seen', {}), session, **kwargs) await self.call_info_reactions(data.get('reactions', {}), session, **kwargs) await self.call_info_comments(data.get('comments', {}), session, **kwargs) async def call_info_views(self, data, session, **kwargs): if 'data' in data: for item in data['data']: view = View() self.set_author(item, view, 'seen') kwargs.get('post').seen['data'].append(view) next_page = data.get('paging', {}).get('next') if next_page is not None: kwargs['recursion'] += 1 new_data = await self.extractor.fetch_url(next_page, session, 'GRAPH', **kwargs) await self.call_info_views(new_data, session, **kwargs) async def call_info_reactions(self, data, session, **kwargs): if 'data' in data: for item in data['data']: reaction = Reaction(item) self.set_author(item, reaction, 'reactions') kwargs.get('post').reactions['data'].append(reaction) next_page = data.get('paging', {}).get('next') if next_page is not None: kwargs['recursion'] += 1 new_data = await self.extractor.fetch_url(next_page, session, 'GRAPH', **