<jupyter_start><jupyter_text>Welcome to the seoml template. > Example Template to pull, store, visualize and transform data into various deliverables.This Notebook contains common SEO jobs such as pulling all your data from Google Search Console [GSC](https://search.google.com/search-console/about) and tracking search engine result pages (SERPs). SQLite is used to store the data locally. Learn from and customize it for your needs. It is an example of using Jupyter Notebook as an exploratory environment and "extracting" finished jobs into fully automatable packages. Learn more about [seoml](https://github.com/miklevin/seoml/blob/main/README.md). Instructions1. If running from Jupyter for the 1st time, change ***site*** below to one of your sites.1. Run the code-blocks in this template from the top-down. 1. Customize to your own needs. NEXT STEP:  The following code-block does these things:1. Imports all the "global" packages.1. Sets a default argument values (if run for the first time from Jupyter).1. Sets up logging to send output both here and to rotating giles in ./logs folder.1. Creates tiny global helper functions using lambda one-liners.1. Detects the context from which script is run: Jupyter vs. Command-line vs. import.1. Parses input when run from the command-line, or uses default or previously-used arguments. 1. Tests that args have been acquired. Run next step.<jupyter_code># export

# Import all globals
import pickle
import argparse
from pathlib import Path
from pprint import pprint
from datetime import date
from pyfiglet import Figlet
import logging, logging.handlers

site = "mikelev.in"  # Set a default argument values (if run for the first time from Jupyter).

# Set up logging to send output both here and to rotating giles in ./logs folder.
ldr = Path("./logs")
if not ldr.exists():
    Path.mkdir(ldr)
dft = "%Y-%m-%dT%H:%M:%S"
fmt = "%(asctime)s %(levelname)s %(name)s.%(funcName)s:%(lineno)d %(message)s"
rfh = logging.handlers.RotatingFileHandler
hlr = [rfh(ldr / "log.txt", maxBytes=100000, backupCount=10), logging.StreamHandler()]
logging.basicConfig(level=logging.INFO, handlers=hlr, format=fmt, datefmt=dft)
logger = logging.getLogger(__name__)

# Create tiny global helper functions using lambda one-liners.
fig = lambda x: print(Figlet(font="standard", width=200).renderText(x).rstrip())
log = lambda x: logger.info(x)
pkl = lambda x: pickle.dumps(x)
unpkl = lambda x: pickle.loads(x)

# Detect the context from which script is run: Jupyter vs. Command-line vs. import.
if hasattr(__builtins__, "__IPYTHON__"):
    # If run from Jupyter, use default or previously-used arguments.
    from IPython.display import display, Markdown

    mkd = lambda x: display(Markdown(x))
    try:
        with open("./seoml/args.pkl", "rb") as handle:
            args = pickle.load(handle)
        log(f"Run from Jupyter, args from Pickle: {args}")
    except:
        args = argparse.Namespace()
        args.site = site
        log(f"Run from Jupyter, args from Global: {args}")
elif __name__ == "__main__":
    # If run from Command-Line, parse input.
    parser = argparse.ArgumentParser()
    add_arg = parser.add_argument
    mkd = lambda x: print(x)
    add_arg("-n", "--site", required=True)
    args = parser.parse_args()
    with open("args.pkl", "wb") as handle:
        pickle.dump(args, handle)
    log(f"Run from Terminal, args from Command-Line: {args}")
else:
    # If imported as a module, use default or previously-used arguments.
    mkd = lambda x: print(x)
    try:
        with open("./args.pkl", "rb") as handle:
            args = pickle.load(handle)
        log(f"Run from Import, args from Pickle: {args}")
    except:
        args = argparse.Namespace()
        args.site = site
        log(f"Run from Import, args from Global: {args}")

assert type(args) == argparse.Namespace  # Test that args have been acquired.

# Create h1 - h6 beautification functions
[exec(z, globals()) for z in [f"h{x} = lambda y: mkd('{'#'*x} %s' % y)" for x in range(1, 7)]]

h1("Create globals & helper functions, parse args.")
fig("Setup Complete!")
h3(f"site={site}")
print("(Change it to one of yours.)") if site == "mikelev.in" else print("Good job!")
h2("Done!")<jupyter_output><empty_output><jupyter_text>--- NEXT STEP: Create Data-Collection Keys.We are about to make requests of some system (GSC) using some arguments (dates, dimensions). So long as these arguments can be expressed as strings, they can also be used as the keys in a key/value database to retreive the data again locally. With this technique, we can retreive all the data we need and store it fast without worrying about the rows and columns of tabular data (like Excel or SQL).Python's namedtuples are perfect here because they are hashable and immutable (what keys need to be). So we can use the same arguments we used to origianlly fetch the data to again retreive it locally. It's no mistake if this sounds like a cache to you. The technique's called memoization, and is found everywhere. The closer your custom datatype is to your actual API-call, the easier it will be. The following code-block does these things:1. Imports the namedtuple factory function from collections.1. Creates a subclass (Gurls) from the namedtuple factory function.1. Creates an instance of the subclass (gurls) by providing arguments. Run next step.<jupyter_code>h1("Create Data-Collection Keys using nametuples.")
from collections import namedtuple

print("https://docs.python.org/3/library/collections.html#collections.namedtuple")
Gurls = namedtuple("Gurls", "startDate, endDate, dimensions, rowLimit, startRow")  # Create subclass
gurls = Gurls("2020-01-01", "2020-01-01", ["page"], 5000, 0)  # Create instance

print(f"gurls = {gurls}")
print(f"Gurls is {type(Gurls)} while gurls is {type(gurls)}.")

h2("Done!")<jupyter_output><empty_output><jupyter_text>--- NEXT STEP: Setup Persistent Dictionary.Key/value pairs is a very tranditional way to do simple databases (that don't have rows & columns like SQL or Excel). Python's built-in dict datatype is actually a tiny non-persistent database. So all we need to do is make it persistent, meaning that the data is still there the next time we run the script. This gives us a place to rapidly "dump" raw data as we collect it. Client/server databases like Redis or memcached are used for this "at scale". But for our use, serverless Sqlite is sufficient.This first delete_me.db database is just to show you how to use the standard Python dict API as a persistent dictionary. The dict becomes persistent through the use of the "with sqldict() as dictname" context manager API pattern. In other words, "dictname" is created after the "as" of "with open" and can only be used within that with's indent. An extra dictname.commit() is also required to write the data to disk (making it persistent).Reads a row of data back out of database. The following code-block does these things:1. Imports the sqlitedict 3rd-party package (you must install sqlitedict).1. Creates its own subfolder (using site name) in which to store the data. 1. Creates an example database.1. Writes a row of data into the database.1. Reads a row of data back out of database. Run next step.<jupyter_code>h1("Setup Persistent Dictionary using SqliteDict.")
from sqlitedict import SqliteDict as sqldict

db_name = "delete_me.db"

# Pick either the single cache or the daily cache for the cache that's used.
single_cache = Path(f"./cache/{site}")
daily_cache = Path(f"./cache/{site}-{date.today()}")
cache_used = single_cache  # Pick one.
if not cache_used.exists():
    Path.mkdir(cache_used, parents=True)  # Create subfolder

h3(f"File Location: {cache_used}")
sample_data = f"When key is {gurls}, value is this message. I could be a big blob."

h3("Data goes in:")
with sqldict(Path(cache_used / db_name)) as db:  # Create example database
    # This looks like a good place for a loop.
    key = <KEY>)
    db[key] = sample_data  #  Write a row of data into the database
    db.commit()
    print(f"{db[key]}")

h3("Data comes out:")
with sqldict(Path(cache_used / db_name)) as db:
    for i, key in enumerate(db):
        print(f"{i + 1} of {len(db)}: {db[key]}")  # Read a row of data back out of database.
        break
    assert len(db) > 0

h2("Done!")<jupyter_output><empty_output><jupyter_text>--- NEXT STEP: Build a list of API-calls.The above step of creating a custom namedtuple gave us the building-block we need to create a list of every call we need to make to the Google Search Console API. Conceptually, we are treating the list of Gurls that will eventually pull the data as being just as important as the data that gets pulled. These "arguments" to the API are actually as important as the data itself, as they will work as the "keys" to later pull the data and a means of ensuring there is no missing gaps in the data. It is closely related to the database concept of Primary Keys.To accomplish this, we create nested loops. One outer loop spins through each day of the last 16 months, while the inner loop spins through 10 "steps" that must be taken through each day to ensure that the maximum number of pages come back for the site for each day in GSC. The following code-block does these things:1. Imports the date-datatype and date-adding functions.1. Creates list of api-friendly string dates.1. Creates step-by intervals needed for each day.1. Create a list of gurls object by comining each day with step.1. Shows the first 30 items in the list. Run next step.<jupyter_code>h1("Build a list of all the API data-calls to be made (and used as keys).")
from datetime import date, timedelta as td
from dateutil.relativedelta import relativedelta as rd

months = 16  # GSC data typically available

# Create list of api-friendly string dates.
today = date.today()
date_diff = today - (today - rd(months=months))
dates = [f"{today - td(days=x + 1)}" for x in range(0, date_diff.days)]  # Create a list of dates
h3(f"Here's first and last 4 days of last {months} months ({len(dates)} days total):")
print(f"Recent: {dates[0:4]}...")
print(f"Oldest: {dates[-4:]}")

# Create step-by intervals needed for each day.
step_by = 5000
total_steps = 10
start_rows = [x for x in range(0, step_by * 10, step_by)]
h3(f'Next we need {total_steps} "start rows" for paging through "chunked" data:')
print(f"{start_rows}")

# Create a list of gurls object by comining each day with step.
h3("Finally we populate a list with Gurls to be our DB-keys and API-calls (showing first 30).")
lot_gurls = []
for i, _date in enumerate(dates):
    for _start in start_rows:
        gurls = Gurls(_date, _date, ["page"], step_by, _start + 1)
        lot_gurls.append(gurls)

pprint(lot_gurls[0:30])

h2("Done!")<jupyter_output><empty_output><jupyter_text>--- NEXT STEP: Rearrange the list for breaking out of days.This step is a list comprehension to turn flat-list structure into nesteed-list structure. Each sub-list will represent a single day with each of its steps. The reason for this transformation is so that once we encounter a step within the day that doesn't return any data, we can break out of that day and immdiately move onto the next knowing that we have already retreived the maximum number of URLs for that day. Attempts to break without this nested list structure would simply go to the next item in the list and likely belong to the same day. The following code-block does these things:1. Group the days into sub-lists, each containing 1 day with all its steps.1. Shows the first 3 days as an example. Run next step.<jupyter_code>h1("Group the days into sub-lists, each containing 1 day with all its steps.")

lot_days = [lot_gurls[x : x + 10] for x in range(0, len(lot_gurls), 10)]
lot_days.reverse()

h3("Showing first 3 day-groups.")
pprint(lot_days[:3])

h2("Done!")<jupyter_output><empty_output><jupyter_text>--- NEXT STEP: Get Google OAuth2 login credentials for Google services.This is often the biggest barrier to noobs, so I wrote the ohawf package to handle it for you. If you haven't already, **pip instal ohawf**. When this is run for the first time it will begin the Google OAuth2 prompt processs. You click on or copy/paste the long URL into a browser. Choose which account you want to connect as. This will control what Google Search Console sites you can access. Refer to https://pypi.org/project/ohawf/ for more details. The following code-block does these things:1. Imports packages to allow Google login and connecting to google services.1. Logs you in or refreshes authentication token.1. Connects to Google Search Console to list owned sites Run next step.<jupyter_code>h1("Get Google OAuth2 login credentials for Google services.")
import ohawf
from apiclient.discovery import build

credentials = ohawf.Credentials().get()  # Log you in or refreshes authentication token.
log(credentials)

h3("Connecting to Google Search Console to list owned sites:")
gsc_service = build("webmasters", "v3", credentials=credentials, cache_discovery=False)
gsc_sites = gsc_service.sites().list().execute()
[print(x["siteUrl"]) for x in gsc_sites["siteEntry"]]

h2("Done!")<jupyter_output><empty_output><jupyter_text>--- NEXT STEP: Get all URLs from site.This step loops through all the API request objects  The following code-block does these things:1. Creates get_urls() helper-function to execute queries against GSC.1. Defines what an "empty" API-response looks like (contains no rows).1. Creates a database to store responses to API-calls.1. Steps through each day, skipping ones that already have data.1. Fetches data for each day, step by step. Breaks to next day on empty response. Run next step.<jupyter_code>h1("Get all URLs from site.")

get_urls = lambda s, r: gsc_service.searchanalytics().query(siteUrl=s, body=r).execute()

empty_response = {"responseAggregationType": "byPage"}  # An api-response with no rows.
db_name = "gurls.db"

with sqldict(Path(cache_used / db_name)) as db:  # Creates database
    h3("Dots (.) represent cached-data and exclamation (!) represents api-hits.")
    print("Day:", end="")
    for i, day in enumerate(lot_days):
        print(f" {len(lot_days) - i}", end="")
        for step, gurls in enumerate(day):
            key = <KEY>)
            if key in db:  # Skips if data already collected
                print(f".", end="")
                break
            else:
                response = get_urls(site, lot_days[i][step]._asdict())  # Fetch data
                if response == empty_response:
                    break
                else:
                    print("!", end="")
                    db[key] = response
                    db.commit()
print("Done!")<jupyter_output><empty_output><jupyter_text>--- NEXT STEP: Get all URLs from site. The following code-block does these things:1.  Run next step.<jupyter_code>h3("Data comes out:")
with sqldict(Path(cache_used / db_name)) as db:
    for i, key in enumerate(db):
        # print(f"{i + 1} of {len(db)}: {db[key]}")
        print(f"{i + 1} of {len(db)} size: {len(db[key])}")
        last = db[key]
        break<jupyter_output><empty_output><jupyter_text>--- NEXT STEP: Get all URLs from site. The following code-block does these things:1.  Run next step.<jupyter_code>all_urls = set()
with sqldict(Path(cache_used / db_name)) as db:
    for key in db:
        rec = db[key]
        if "rows" in rec:
            all_urls.update({x["keys"][0] for x in rec["rows"]})
        print(".", end="")
import sqlite3

sql = """CREATE TABLE IF NOT EXISTS url (
  url TEXT NOT NULL PRIMARY KEY
) WITHOUT ROWID;"""

connection = sqlite3.connect(Path(cache_used / "urls.db"))
cursor = connection.cursor()
cursor.execute(sql)
connection.commit()
for url in all_urls:
    try:
        cursor.execute(f'INSERT INTO url (url) VALUES ("{url}");')
    except:
        pass
connection.commit()
cursor.close()
connection.close()
connection = sqlite3.connect(Path(cache_used / "urls.db"))
cursor = connection.cursor()
records = cursor.execute("SELECT url FROM url LIMIT 1000;")
for i, record in enumerate(records):
    print(i + 1, record[0])
cursor.close()
connection.close()
import sqlite3

sql = """CREATE TABLE IF NOT EXISTS url_history (
  date TEXT,
  url TEXT,
  clicks INTEGER,
  ctr FLOAT,
  impressions FLOAT,
  position FLOAT,
  PRIMARY KEY (date, url)
) WITHOUT ROWID;"""

connection = sqlite3.connect(Path(cache_used / "url_history.db"))
cursor = connection.cursor()
cursor.execute(sql)
connection.commit()

with sqldict(Path(cache_used / db_name)) as db:
    for key in db:
        rec = db[key]
        if "rows" in rec:
            rows = rec["rows"]
            for row in rows:
                vals = [row[x] if i else f"'{row[x][0]}'" for i, x in enumerate(row)]
                vals = [f"'{unpkl(key).startDate}'"] + vals
                vals = ", ".join([f"{x}" for x in vals])
                stmt = f"INSERT INTO url_history VALUES ({vals});"
                try:
                    cursor.execute(stmt)
                except:
                    continue
            connection.commit()
            print(".", end="")
cursor.close()
connection.close()
import pandas as pd

connection = sqlite3.connect(Path(cache_used / "url_history.db"))
df = pd.read_sql_query("SELECT * FROM url_history", connection)
connection.close()
df
pip install pycausalimpact<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.10123371382451285, 'fraction_numerical': 0.009166378415773089, 'mean_word_length': 3.8360747142458878, 'pattern_counts': {'":': 2, '<': 39, '<?xml version=': 0, '>': 41, 'https://': 4, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '9', 'dir': 'jupyter-structured-clean-dedup', 'id': '36639', 'license': 'Apache-2.0', 'n_tokens_mistral': 5304, 'n_tokens_neox': 5163, 'n_words': 2169, 'path': '00_core.ipynb', 'repo_name': 'miklevin/seoml', 'size': '17354'}