Twitter

from obsei.source.twitter_source import TwitterCredentials, TwitterSource, TwitterSourceConfig

# 初始化 Twitter 源配置
source_config = TwitterSourceConfig(
   keywords=["issue"], # 关键词、@用户或#话题标签
   lookup_period="1h", # 从当前时间开始的查找周期，格式：<数字><d|h|m>（天|小时|分钟）
   cred_info=TwitterCredentials(
       # 输入您的 Twitter 消费者密钥和秘密。可在 https://developer.twitter.com/en/apply-for-access 获取
       consumer_key="<twitter_consumer_key>",
       consumer_secret="<twitter_consumer_secret>",
       bearer_token='<请输入 Bearer Token>',
   )
)

# 初始化推文检索器
source = TwitterSource()

YouTube 抓取器

from obsei.source.youtube_scrapper import YoutubeScrapperSource, YoutubeScrapperConfig

# 初始化 YouTube 源配置
source_config = YoutubeScrapperConfig(
    video_url="https://www.youtube.com/watch?v=uZfns0JIlFk", # YouTube 视频 URL
    fetch_replies=True, # 获取评论回复
    max_comments=10, # 要获取的评论和回复总数
    lookup_period="1Y", # 从当前时间开始的查找周期，格式：<数字><d|h|m|M|Y>（天|小时|分钟|月|年）
)

# 初始化 YouTube 评论检索器
source = YoutubeScrapperSource()

Facebook

from obsei.source.facebook_source import FacebookCredentials, FacebookSource, FacebookSourceConfig

# 初始化 Facebook 源配置
source_config = FacebookSourceConfig(
   page_id="110844591144719", # Facebook 页面 ID，例如 Obsei 的这个页面
   lookup_period="1h", # 从当前时间开始的查找周期，格式：<数字><d|h|m>（天|小时|分钟）
   cred_info=FacebookCredentials(
       # 输入您的 Facebook 应用 ID、应用秘密和长期令牌。可在 https://developers.facebook.com/apps/ 获取
       app_id="<facebook_app_id>",
       app_secret="<facebook_app_secret>",
       long_term_token="<facebook_long_term_token>",
   )
)

# 初始化 Facebook 帖子评论检索器
source = FacebookSource()

电子邮件

from obsei.source.email_source import EmailConfig, EmailCredInfo, EmailSource

# 初始化电子邮件源配置
source_config = EmailConfig(
   # 最常用电子邮件提供商的 IMAP 服务器列表
   # https://www.systoolsgroup.com/imap/
   # 另外，如果您使用的是 Gmail 账户，请确保在您的账户中允许不太安全的应用程序 -
   # https://myaccount.google.com/lesssecureapps?pli=1
   # 同时启用 IMAP 访问 -
   # https://mail.google.com/mail/u/0/#settings/fwdandpop
   imap_server="imap.gmail.com", # 输入 IMAP 服务器
   cred_info=EmailCredInfo(
       # 输入您的电子邮件账户用户名和密码
       username="<email_username>",
       password="<email_password>"
   ),
   lookup_period="1h" # 从当前时间开始的查找周期，格式：<数字><d|h|m>（天|小时|分钟）
)

# 初始化电子邮件检索器
source = EmailSource()

Google 地图评论抓取器

from obsei.source.google_maps_reviews import OSGoogleMapsReviewsSource, OSGoogleMapsReviewsConfig

# 初始化Outscrapper Maps评论源配置
source_config = OSGoogleMapsReviewsConfig(
   # 从https://outscraper.com/获取API密钥
   api_key="<输入您的API密钥>",
   # 输入Google地图链接或地点ID
   # 例如，以下为“泰姬陵”的链接
   queries=["https://www.google.co.in/maps/place/Taj+Mahal/@27.1751496,78.0399535,17z/data=!4m5!3m4!1s0x39747121d702ff6d:0xdd2ae4803f767dde!8m2!3d27.1751448!4d78.0421422"],
   number_of_reviews=10,
)


# 初始化Outscrapper Maps评论获取器
source = OSGoogleMapsReviewsSource()

AppStore评论抓取器

from obsei.source.appstore_scrapper import AppStoreScrapperConfig, AppStoreScrapperSource

# 初始化AppStore源配置
source_config = AppStoreScrapperConfig(
   # 需要两个参数：app_id和country。
   # `app_id`可在App Store应用URL的末尾找到。
   # 例如 - https://apps.apple.com/us/app/xcode/id497799835
   # `310633997`是Xcode的app_id，`us`是国家。
   countries=["us"],
   app_id="310633997",
   lookup_period="1h" # 查找周期，从当前时间开始，格式：<数字><d|h|m>（天|小时|分钟）
)


# 初始化AppStore评论获取器
source = AppStoreScrapperSource()

Play Store评论抓取器

from obsei.source.playstore_scrapper import PlayStoreScrapperConfig, PlayStoreScrapperSource

# 初始化Play Store源配置
source_config = PlayStoreScrapperConfig(
   # 需要两个参数：package_name和country。
   # `package_name`可在Play Store应用URL的末尾找到。
   # 例如 - https://play.google.com/store/apps/details?id=com.google.android.gm&hl=en&gl=US
   # `com.google.android.gm`是Xcode的package_name，`us`是国家。
   countries=["us"],
   package_name="com.google.android.gm",
   lookup_period="1h" # 查找周期，从当前时间开始，格式：<数字><d|h|m>（天|小时|分钟）
)

# 初始化Play Store评论获取器
source = PlayStoreScrapperSource()

Reddit

from obsei.source.reddit_source import RedditConfig, RedditSource, RedditCredInfo

# 初始化Reddit源配置
source_config = RedditConfig(
   subreddits=["wallstreetbets"], # 子Reddit列表
   # Reddit账号用户名和密码
   # 也可以输入Reddit client_id、client_secret或refresh_token
   # 在https://www.reddit.com/prefs/apps创建凭据
   # 参考https://praw.readthedocs.io/en/latest/getting_started/authentication.html
   # 目前支持密码流、只读模式和已保存刷新令牌模式
   cred_info=RedditCredInfo(
       username="<reddit_username>",
       password="<reddit_password>"
   ),
   lookup_period="1h" # 查找周期，从当前时间开始，格式：<数字><d|h|m>（天|小时|分钟）
)

# 初始化Reddit获取器
source = RedditSource()

Reddit抓取器

注意：Reddit对抓取器有严格的速率限制，因此请在长时间内使用它来获取少量数据

from obsei.source.reddit_scrapper import RedditScrapperConfig, RedditScrapperSource

# 初始化Reddit抓取器源配置
source_config = RedditScrapperConfig(
   # Reddit子Reddit、搜索等RSS URL。如需正确URL，请参考以下链接：
   # 参考https://www.reddit.com/r/pathogendavid/comments/tv8m9/pathogendavids_guide_to_rss_and_reddit/
   url="https://www.reddit.com/r/wallstreetbets/comments/.rss?sort=new",
   lookup_period="1h" # 查找周期，从当前时间开始，格式：<数字><d|h|m>（天|小时|分钟）
)

# 初始化Reddit获取器
source = RedditScrapperSource()

Google新闻

from obsei.source.google_news_source import GoogleNewsConfig, GoogleNewsSource

# 初始化Google新闻源配置
source_config = GoogleNewsConfig(
   query='bitcoin',
   max_results=5,
   # 要获取全文，启用`fetch_article`标志
   # 默认情况下，Google新闻提供标题和亮点
   fetch_article=True,
   # proxy='http://127.0.0.1:8080'
)

# 初始化Google新闻获取器
source = GoogleNewsSource()

网络爬虫

from obsei.source.website_crawler_source import TrafilaturaCrawlerConfig, TrafilaturaCrawlerSource

# 初始化网站爬虫源配置
source_config = TrafilaturaCrawlerConfig(
   urls=['https://obsei.github.io/obsei/']
)

# 初始化网站文本获取器
source = TrafilaturaCrawlerSource()

Pandas数据框

import pandas as pd
from obsei.source.pandas_source import PandasSource, PandasSourceConfig

# 从CSV、Excel、SQL等来源初始化你的Pandas数据框
# 在以下示例中，我们读取包含两列的CSV文件：title和text
csv_file = "https://raw.githubusercontent.com/deepset-ai/haystack/master/tutorials/small_generator_dataset.csv"
dataframe = pd.read_csv(csv_file)

# 初始化Pandas数据源配置
sink_config = PandasSourceConfig(
   dataframe=dataframe,
   include_columns=["score"],
   text_columns=["name", "degree"],
)

# 初始化 Pandas 接收器
sink = PandasSource()

文本分类

文本分类：将文本分类到用户提供的类别中。

from obsei.analyzer.classification_analyzer import ClassificationAnalyzerConfig, ZeroShotClassificationAnalyzer

# 初始化分类分析器配置
# 如果添加了“positive”和“negative”标签，它也可以检测情感。
analyzer_config=ClassificationAnalyzerConfig(
   labels=["服务", "延误", "性能"],
)

# 初始化分类分析器
# 支持的模型请参考 https://huggingface.co/models?filter=zero-shot-classification
text_analyzer = ZeroShotClassificationAnalyzer(
   model_name_or_path="typeform/mobilebert-uncased-mnli",
   device="auto"
)

情感分析器

情感分析器：检测文本的情感。文本分类也可以进行情感分析，但如果您不想使用重型 NLP 模型，则可以使用资源消耗较少的基于词典的 Vader 情感检测器。

from obsei.analyzer.sentiment_analyzer import VaderSentimentAnalyzer

# Vader 不需要任何配置设置
analyzer_config=None

# 初始化 Vader 情感分析器
text_analyzer = VaderSentimentAnalyzer()

NER 分析器

NER（命名实体识别）分析器：从文本中提取信息，并将提到的命名实体分类到预定义的类别中，例如人名、组织、地点、医疗代码、时间表达式、数量、货币值、百分比等。

from obsei.analyzer.ner_analyzer import NERAnalyzer

# NER 分析器不需要配置设置
analyzer_config=None

# 初始化 NER 分析器
# 支持的模型请参考 https://huggingface.co/models?filter=token-classification
text_analyzer = NERAnalyzer(
   model_name_or_path="elastic/distilbert-base-cased-finetuned-conll03-english",
   device = "auto"
)

翻译器

from obsei.analyzer.translation_analyzer import TranslationAnalyzer

# 翻译器不需要分析器配置
analyzer_config = None

# 初始化翻译器
# 支持的模型请参考 https://huggingface.co/models?pipeline_tag=translation
analyzer = TranslationAnalyzer(
   model_name_or_path="Helsinki-NLP/opus-mt-hi-en",
   device = "auto"
)

PII 匿名化器

from obsei.analyzer.pii_analyzer import PresidioEngineConfig, PresidioModelConfig, \
   PresidioPIIAnalyzer, PresidioPIIAnalyzerConfig

# 初始化 PII 分析器的配置
analyzer_config = PresidioPIIAnalyzerConfig(
   # 是否仅返回 PII 分析结果或匿名化文本
   analyze_only=False,
   # 是否返回匿名化决策的详细信息
   return_decision_process=True
)

# 初始化 PII 分析器
analyzer = PresidioPIIAnalyzer(
   engine_config=PresidioEngineConfig(
       # 支持 spacy 和 stanza NLP 引擎
       # 更多信息请参考
       # https://microsoft.github.io/presidio/analyzer/developing_recognizers/#utilize-spacy-or-stanza
       nlp_engine_name="spacy",
       # 更新所需的 spacy 模型和语言
       models=[PresidioModelConfig(model_name="en_core_web_lg", lang_code="en")]
   )
)

虚拟分析器

虚拟分析器：什么都不做。它只是用于将输入（TextPayload）转换为输出（TextPayload），并添加用户提供的虚拟数据。

from obsei.analyzer.dummy_analyzer import DummyAnalyzer, DummyAnalyzerConfig

# 初始化虚拟分析器的配置设置
analyzer_config = DummyAnalyzerConfig()

# 初始化虚拟分析器
analyzer = DummyAnalyzer()

Slack

from obsei.sink.slack_sink import SlackSink, SlackSinkConfig

# 初始化 Slack 接收器配置
sink_config = SlackSinkConfig(
   # 提供 Slack 机器人/应用令牌
   # 更多详情请参考 https://slack.com/intl/en-de/help/articles/215770388-Create-and-regenerate-API-tokens
   slack_token="<Slack_app_token>",
   # 获取频道 ID 请参考 https://stackoverflow.com/questions/40940327/what-is-the-simplest-way-to-find-a-slack-team-id-and-a-channel-id
   channel_id="C01LRS6CT9Q"
)

# 初始化 Slack 接收器
sink = SlackSink()

Zendesk

from obsei.sink.zendesk_sink import ZendeskSink, ZendeskSinkConfig, ZendeskCredInfo

# 初始化 Zendesk 接收器配置
sink_config = ZendeskSinkConfig(
   # 提供 Zendesk 域名
   domain="zendesk.com",
   # 如果您有子域名，请提供
   subdomain=None,
   # 输入 Zendesk 用户详细信息
   cred_info=ZendeskCredInfo(
       email="<zendesk_user_email>",
       password="<zendesk_password>"
   )
)

# 初始化 Zendesk 接收器
sink = ZendeskSink()

Jira

from obsei.sink.jira_sink import JiraSink, JiraSinkConfig

# 为了测试，您可以本地启动 Jira 服务器
# 参考 https://developer.atlassian.com/server/framework/atlassian-sdk/atlas-run-standalone/

# 初始化 Jira 接收器配置
sink_config = JiraSinkConfig(
   url="http://localhost:2990/jira", # Jira 服务器 URL
    # 具有创建问题权限的用户的用户名和密码
   username="<username>",
   password="<password>",
   # 要创建的问题类型
   # 更多信息请参考 https://support.atlassian.com/jira-cloud-administration/docs/what-are-issue-types/
   issue_type={"name": "Task"},
   # 在哪个项目下创建问题
   # 更多信息请参考 https://support.atlassian.com/jira-software-cloud/docs/what-is-a-jira-software-project/
   project={"key": "CUS"},
)

# 初始化 Jira 接收器
sink = JiraSink()

ElasticSearch

from obsei.sink.elasticsearch_sink import ElasticSearchSink, ElasticSearchSinkConfig

# 为了测试，您可以使用 Docker 本地启动 Elasticsearch 服务器
# `docker run -d --name elasticsearch -p 9200:9200 -e "discovery.type=single-node" elasticsearch:8.5.0`

# 初始化 Elasticsearch 接收器配置
sink_config = ElasticSearchSinkConfig(
   # Elasticsearch 服务器
   hosts="http://localhost:9200",
   # 索引名称，如果不存在则会自动创建
   index_name="test",
)

# 初始化 Elasticsearch 接收器
sink = ElasticSearchSink()

Http

from obsei.sink.http_sink import HttpSink, HttpSinkConfig

# 为了测试，您可以使用 Postman 创建一个模拟 HTTP 服务器
# 更多详情请参考 https://learning.postman.com/docs/designing-and-developing-your-api/mocking-data/setting-up-mock/

# 初始化 HTTP 接收器配置（目前仅支持 POST 请求）
sink_config = HttpSinkConfig(
   # 提供 HTTP 服务器 URL
   url="https://localhost:8080/api/path",
   # 您可以在这里添加希望随请求一起发送的头部信息
   headers={
       "Content-type": "application/json"
   }
)

# 如果需要修改或转换负载数据，可以创建转换器类
# 请参考 obsei.sink.dailyget_sink.PayloadConvertor 示例

# 初始化 HTTP 接收器
sink = HttpSink()

Pandas DataFrame

from pandas import DataFrame
from obsei.sink.pandas_sink import PandasSink, PandasSinkConfig

# 初始化 Pandas 接收器配置
sink_config = PandasSinkConfig(
   dataframe=DataFrame()
)

# 初始化 Pandas 接收器
sink = PandasSink()

Logger

这在测试和流水线的干运行中非常有用。

from obsei.sink.logger_sink import LoggerSink, LoggerSinkConfig
import logging
import sys

logger = logging.getLogger("Obsei")
logging.basicConfig(stream=sys.stdout, level=logging.INFO)

# 初始化 logger 接收器配置
sink_config = LoggerSinkConfig(
   logger=logger,
   level=logging.INFO
)

# 初始化 logger 接收器
sink = LoggerSink()