Youtube Scrapper

```python from obsei.source.youtube_scrapper import YoutubeScrapperSource, YoutubeScrapperConfig

initialize Youtube source config

sourceconfig = YoutubeScrapperConfig( videourl="https://www.youtube.com/watch?v=uZfns0JIlFk", # Youtube video URL fetchreplies=True, # Fetch replies to comments maxcomments=10, # Total number of comments and replies to fetch lookup_period="1Y", # Lookup period from current time, format: <number><d|h|m|M|Y> (day|hour|minute|month|year) )

initialize Youtube comments retriever

source = YoutubeScrapperSource() ```

Facebook

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

initialize facebook source config

sourceconfig = FacebookSourceConfig( pageid="110844591144719", # Facebook page id, for example this one for Obsei lookupperiod="1h", # Lookup period from current time, format: <number><d|h|m> (day|hour|minute) credinfo=FacebookCredentials( # Enter your facebook appid, appsecret and longtermtoken. Get it from https://developers.facebook.com/apps/ appid="<facebookappid>", appsecret="", longtermtoken="", ) )

initialize facebook post comments retriever

source = FacebookSource() ```

Email

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

initialize email source config

sourceconfig = EmailConfig( # List of IMAP servers for most commonly used email providers # https://www.systoolsgroup.com/imap/ # Also, if you're using a Gmail account then make sure you allow less secure apps on your account - # https://myaccount.google.com/lesssecureapps?pli=1 # Also enable IMAP access - # https://mail.google.com/mail/u/0/#settings/fwdandpop imapserver="imap.gmail.com", # Enter IMAP server credinfo=EmailCredInfo( # Enter your email account username and password username="<emailusername>", password="" ), lookup_period="1h" # Lookup period from current time, format: <number><d|h|m> (day|hour|minute) )

initialize email retriever

source = EmailSource() ```

Google Maps Reviews Scrapper

```python from obsei.source.googlemapsreviews import OSGoogleMapsReviewsSource, OSGoogleMapsReviewsConfig

initialize Outscrapper Maps review source config

sourceconfig = OSGoogleMapsReviewsConfig( # Collect API key from https://outscraper.com/ apikey="", # Enter Google Maps link or place id # For example below is for the "Taj Mahal" 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"], numberofreviews=10, )

initialize Outscrapper Maps review retriever

source = OSGoogleMapsReviewsSource() ```

AppStore Reviews Scrapper

```python from obsei.source.appstore_scrapper import AppStoreScrapperConfig, AppStoreScrapperSource

initialize app store source config

sourceconfig = AppStoreScrapperConfig( # Need two parameters appid and country. # app_id can be found at the end of the url of app in app store. # For example - https://apps.apple.com/us/app/xcode/id497799835 # 310633997 is the appid for xcode and us is country. countries=["us"], appid="310633997", lookup_period="1h" # Lookup period from current time, format: <number><d|h|m> (day|hour|minute) )

initialize app store reviews retriever

source = AppStoreScrapperSource() ```

Play Store Reviews Scrapper

```python from obsei.source.playstore_scrapper import PlayStoreScrapperConfig, PlayStoreScrapperSource

initialize play store source config

sourceconfig = PlayStoreScrapperConfig( # Need two parameters packagename and country. # package_name can be found at the end of the url of app in play store. # For example - https://play.google.com/store/apps/details?id=com.google.android.gm&hl=en&gl=US # com.google.android.gm is the packagename for xcode and us is country. countries=["us"], packagename="com.google.android.gm", lookup_period="1h" # Lookup period from current time, format: <number><d|h|m> (day|hour|minute) )

initialize play store reviews retriever

source = PlayStoreScrapperSource() ```

Reddit

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

initialize reddit source config

sourceconfig = RedditConfig( subreddits=["wallstreetbets"], # List of subreddits # Reddit account username and password # You can also enter reddit clientid and clientsecret or refreshtoken # Create credential at https://www.reddit.com/prefs/apps # Also refer https://praw.readthedocs.io/en/latest/gettingstarted/authentication.html # Currently Password Flow, Read Only Mode and Saved Refresh Token Mode are supported credinfo=RedditCredInfo( username="", password="" ), lookup_period="1h" # Lookup period from current time, format: <number><d|h|m> (day|hour|minute) )

initialize reddit retriever

source = RedditSource() ```

Reddit Scrapper

Note: Reddit heavily rate limit scrappers, hence use it to fetch small data during long period

```python from obsei.source.reddit_scrapper import RedditScrapperConfig, RedditScrapperSource

initialize reddit scrapper source config

sourceconfig = RedditScrapperConfig( # Reddit subreddit, search etc rss url. For proper url refer following link - # Refer https://www.reddit.com/r/pathogendavid/comments/tv8m9/pathogendavidsguidetorssandreddit/ url="https://www.reddit.com/r/wallstreetbets/comments/.rss?sort=new", lookup_period="1h" # Lookup period from current time, format: <number><d|h|m> (day|hour|minute) )

initialize reddit retriever

source = RedditScrapperSource() ```

Google News

```python from obsei.source.googlenewssource import GoogleNewsConfig, GoogleNewsSource

initialize Google News source config

sourceconfig = GoogleNewsConfig( query='bitcoin', maxresults=5, # To fetch full article text enable fetch_article flag # By default google news gives title and highlight fetch_article=True, # proxy='http://127.0.0.1:8080' )

initialize Google News retriever

source = GoogleNewsSource() ```

Web Crawler

```python from obsei.source.websitecrawlersource import TrafilaturaCrawlerConfig, TrafilaturaCrawlerSource

initialize website crawler source config

source_config = TrafilaturaCrawlerConfig( urls=['https://obsei.github.io/obsei/'] )

initialize website text retriever

source = TrafilaturaCrawlerSource() ```

Pandas DataFrame

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

Initialize your Pandas DataFrame from your sources like csv, excel, sql etc

In following example we are reading csv which have two columns title and text

csvfile = "https://raw.githubusercontent.com/deepset-ai/haystack/master/tutorials/smallgeneratordataset.csv" dataframe = pd.readcsv(csv_file)

initialize pandas sink config

sinkconfig = PandasSourceConfig( dataframe=dataframe, includecolumns=["score"], text_columns=["name", "degree"], )

initialize pandas sink

sink = PandasSource() ```

Sentiment Analyzer

Sentiment Analyzer: Detect the sentiment of the text. Text classification can also perform sentiment analysis but if you don't want to use heavy-duty NLP model then use less resource hungry dictionary based Vader Sentiment detector.

```python from obsei.analyzer.sentiment_analyzer import VaderSentimentAnalyzer

Vader does not need any configuration settings

analyzer_config=None

initialize vader sentiment analyzer

text_analyzer = VaderSentimentAnalyzer() ```

NER Analyzer

NER (Named-Entity Recognition) Analyzer: Extract information and classify named entities mentioned in text into pre-defined categories such as person names, organizations, locations, medical codes, time expressions, quantities, monetary values, percentages, etc

```python from obsei.analyzer.ner_analyzer import NERAnalyzer

NER analyzer does not need configuration settings

analyzer_config=None

initialize ner analyzer

For supported models refer https://huggingface.co/models?filter=token-classification

textanalyzer = NERAnalyzer( modelnameorpath="elastic/distilbert-base-cased-finetuned-conll03-english", device = "auto" ) ```

Translator

```python from obsei.analyzer.translation_analyzer import TranslationAnalyzer

Translator does not need analyzer config

analyzer_config = None

initialize translator

For supported models refer https://huggingface.co/models?pipeline_tag=translation

analyzer = TranslationAnalyzer( modelnameor_path="Helsinki-NLP/opus-mt-hi-en", device = "auto" ) ```

PII Anonymizer

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

initialize pii analyzer's config

analyzerconfig = PresidioPIIAnalyzerConfig( # Whether to return only pii analysis or anonymize text analyzeonly=False, # Whether to return detail information about anonymization decision returndecisionprocess=True )

initialize pii analyzer

analyzer = PresidioPIIAnalyzer( engineconfig=PresidioEngineConfig( # spacy and stanza nlp engines are supported # For more info refer # https://microsoft.github.io/presidio/analyzer/developingrecognizers/#utilize-spacy-or-stanza nlpenginename="spacy", # Update desired spacy model and language models=[PresidioModelConfig(modelname="encoreweblg", lang_code="en")] ) ) ```

Dummy Analyzer

Dummy Analyzer: Does nothing. Its simply used for transforming the input (TextPayload) to output (TextPayload) and adding the user supplied dummy data.

```python from obsei.analyzer.dummy_analyzer import DummyAnalyzer, DummyAnalyzerConfig

initialize dummy analyzer's configuration settings

analyzer_config = DummyAnalyzerConfig()

initialize dummy analyzer

analyzer = DummyAnalyzer() ```

Zendesk

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

initialize zendesk sink config

sinkconfig = ZendeskSinkConfig( # provide zendesk domain domain="zendesk.com", # provide subdomain if you have one subdomain=None, # Enter zendesk user details credinfo=ZendeskCredInfo( email="", password="" ) )

initialize zendesk sink

sink = ZendeskSink() ```

Jira

```python from obsei.sink.jira_sink import JiraSink, JiraSinkConfig

For testing purpose you can start jira server locally

Refer https://developer.atlassian.com/server/framework/atlassian-sdk/atlas-run-standalone/

initialize Jira sink config

sinkconfig = JiraSinkConfig( url="http://localhost:2990/jira", # Jira server url # Jira username & password for user who have permission to create issue username="", password="", # Which type of issue to be created # For more information refer https://support.atlassian.com/jira-cloud-administration/docs/what-are-issue-types/ issuetype={"name": "Task"}, # Under which project issue to be created # For more information refer https://support.atlassian.com/jira-software-cloud/docs/what-is-a-jira-software-project/ project={"key": "CUS"}, )

initialize Jira sink

sink = JiraSink() ```

ElasticSearch

```python from obsei.sink.elasticsearch_sink import ElasticSearchSink, ElasticSearchSinkConfig

For testing purpose you can start Elasticsearch server locally via docker

docker run -d --name elasticsearch -p 9200:9200 -e "discovery.type=single-node" elasticsearch:8.5.0

initialize Elasticsearch sink config

sinkconfig = ElasticSearchSinkConfig( # Elasticsearch server hosts="http://localhost:9200", # Index name, it will create if not exist indexname="test", )

initialize Elasticsearch sink

sink = ElasticSearchSink() ```

Http

```python from obsei.sink.http_sink import HttpSink, HttpSinkConfig

For testing purpose you can create mock http server via postman

For more details refer https://learning.postman.com/docs/designing-and-developing-your-api/mocking-data/setting-up-mock/

initialize http sink config (Currently only POST call is supported)

sink_config = HttpSinkConfig( # provide http server url url="https://localhost:8080/api/path", # Here you can add headers you would like to pass with request headers={ "Content-type": "application/json" } )

To modify or converting the payload, create convertor class

Refer obsei.sink.dailyget_sink.PayloadConvertor for example

initialize http sink

sink = HttpSink() ```

Pandas DataFrame

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

initialize pandas sink config

sink_config = PandasSinkConfig( dataframe=DataFrame() )

initialize pandas sink

sink = PandasSink() ```

Logger

This is useful for testing and dry running the pipeline.

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

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

initialize logger sink config

sink_config = LoggerSinkConfig( logger=logger, level=logging.INFO )

initialize logger sink

sink = LoggerSink() ```