123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240 |
- #!/usr/bin/env python
- # -*- coding: utf-8 -*-
- # Copyright 2017 The Chromium Authors. All rights reserved.
- # Use of this source code is governed by a BSD-style license that can be
- # found in the LICENSE file.
- """Fetches articles from the server.
- Examples:
- $ fetch.py # unauthenticated, no experiments
- $ fetch.py --short # abbreviate instead of dumping JSON
- $ fetch.py --signed-in -x3313279 # authenticated, results from Google Now
- If getting signed-in results, authenticates with OAuth2 and stores the
- credentials at ~/.zineauth.
- """
- from __future__ import absolute_import, division, print_function, unicode_literals
- import argparse
- import base64
- import datetime
- import json
- import os
- import textwrap
- import oauth2client.client
- import oauth2client.file
- import oauth2client.tools
- import requests
- import sys
- API_KEY_FILE = os.path.join(
- os.path.dirname(__file__),
- "../../../google_apis/internal/google_chrome_api_keys.h")
- API_SCOPE = "https://www.googleapis.com/auth/chrome-content-suggestions"
- API_HOSTS = {
- "prod": "https://chromecontentsuggestions-pa.googleapis.com",
- "staging": "https://staging-chromecontentsuggestions-pa.googleapis.com",
- "alpha": "https://alpha-chromecontentsuggestions-pa.sandbox.googleapis.com",
- }
- API_PATH = "/v1/suggestions/fetch"
- def main():
- default_lang = os.environ.get("LANG", "en_US").split(".")[0]
- parser = argparse.ArgumentParser(
- description="fetch articles from server",
- parents=[oauth2client.tools.argparser])
- parser.add_argument("-c", "--component",
- default="prod", choices=["prod", "staging", "alpha"],
- help="component to fetch from (default: prod)")
- parser.add_argument("-x", "--experiment", action="append", type=int,
- help="include an experiment ID")
- parser.add_argument("-l", "--ui-language", default=default_lang,
- help="language code (default: %s)" % default_lang)
- parser.add_argument("--ip", help="fake IP address")
- parser.add_argument("--api-key", type=str,
- help="API key to use for unauthenticated requests"
- " (default: use official key)")
- parser.add_argument("-s", "--signed-in", action="store_true",
- help="sign in and issue authenticated request")
- parser.add_argument("--client", metavar="ID,SECRET", type=str,
- help="client project to use for authenticated requests"
- " (default: use official project ID")
- parser.add_argument("--short", action="store_true",
- help="print results in abbreviated form")
- args = parser.parse_args()
- r = PostRequest(args)
- j = {}
- try:
- j = r.json()
- except ValueError:
- print(r.text.encode("utf-8"))
- sys.exit(1)
- if j.get("error"):
- print(r.text.encode("utf-8"))
- sys.exit(1)
- if args.short:
- PrintShortResponse(j)
- return
- print(r.text.encode("utf-8"))
- if r.status_code != 200:
- sys.exit(1)
- def GetApiKeyFile():
- return API_KEY_FILE
- def GetAPIDefs():
- """Parses the internal file with API keys and returns a dict."""
- with open(GetApiKeyFile()) as f:
- lines = f.readlines()
- defs = {}
- next_name = None
- for line in lines:
- if next_name:
- defs[next_name] = json.loads(line)
- next_name = None
- elif line.startswith("#define"):
- try:
- _, name, value = line.split()
- except ValueError:
- continue
- if value == "\\":
- next_name = name
- else:
- defs[name] = json.loads(value)
- return defs
- def GetAPIKey():
- return GetAPIDefs()["GOOGLE_API_KEY"]
- def GetOAuthClient():
- defs = GetAPIDefs()
- return defs["GOOGLE_CLIENT_ID_MAIN"], defs["GOOGLE_CLIENT_SECRET_MAIN"]
- def EncodeExperiments(experiments):
- """Turn a list of experiment IDs into an X-Client-Data header value.
- Encodes all the IDs as a protobuf (tag 1, varint) and base64 encodes the
- result.
- """
- binary = b""
- for exp in experiments:
- binary += b"\x08"
- while True:
- byte = (exp & 0x7f)
- exp >>= 7
- if exp:
- binary += chr(0x80 | byte)
- else:
- binary += chr(byte)
- break
- return base64.b64encode(binary)
- def AbbreviateDuration(duration):
- """Turn a datetime.timedelta into a short string like "10h 14m"."""
- w = duration.days // 7
- d = duration.days % 7
- h = duration.seconds // 3600
- m = (duration.seconds % 3600) // 60
- s = duration.seconds % 60
- us = duration.microseconds
- if w:
- return "%dw %dd" % (w, d)
- elif d:
- return "%dd %dh" % (d, h)
- elif h:
- return "%dh %dm" % (h, m)
- elif m:
- return "%dm %ds" % (m, s)
- elif s:
- return "%ds" % s
- elif us:
- return "<1s"
- else:
- return "0s"
- def PostRequest(args):
- url = API_HOSTS[args.component] + API_PATH
- headers = {}
- if args.experiment:
- headers["X-Client-Data"] = EncodeExperiments(args.experiment)
- if args.ip is not None:
- headers["X-User-IP"] = args.ip
- if args.signed_in:
- if args.client:
- client_id, client_secret = args.client.split(",")
- else:
- client_id, client_secret = GetOAuthClient()
- Authenticate(args, headers, client_id, client_secret)
- else:
- if args.api_key:
- api_key = args.api_key
- else:
- api_key = GetAPIKey()
- url += "?key=" + api_key
- data = {
- "uiLanguage": args.ui_language,
- }
- return requests.post(url, headers=headers, data=data)
- def Authenticate(args, headers, client_id, client_secret):
- storage = oauth2client.file.Storage(os.path.expanduser("~/.zineauth"))
- creds = storage.get()
- if not creds or creds.invalid or creds.access_token_expired:
- flow = oauth2client.client.OAuth2WebServerFlow(
- client_id=client_id, client_secret=client_secret,
- scope=API_SCOPE)
- oauth2client.tools.run_flow(flow, storage, args)
- creds = storage.get()
- creds.apply(headers)
- def PrintShortResponse(j):
- now = datetime.datetime.now()
- for category in j["categories"]:
- print("%s: " % category["localizedTitle"])
- for suggestion in category.get("suggestions", []):
- attribution = suggestion["attribution"]
- title = suggestion["title"]
- full_url = suggestion["fullPageUrl"]
- amp_url = suggestion.get("ampUrl")
- creation_time = suggestion["creationTime"]
- if len(title) > 40:
- title = textwrap.wrap(title, 40)[0] + "…"
- creation_time = ParseDateTime(creation_time)
- age = AbbreviateDuration(now - creation_time)
- print(" “%s” (%s, %s ago)" % (title, attribution, age))
- print(" " + (amp_url or full_url))
- if category["allowFetchingMoreResults"]:
- print(" [More]")
- def ParseDateTime(creation_time):
- try:
- return datetime.datetime.strptime(creation_time, "%Y-%m-%dT%H:%M:%SZ")
- except ValueError:
- return datetime.datetime.strptime(creation_time, "%Y-%m-%dT%H:%M:%S.%fZ")
- if __name__ == "__main__":
- main()
|