Merge pull request #64 from nstrydom2/revision

revision

Author: nstrydom2

.gitignore

@@ -106,6 +106,7 @@ celerybeat.pid
 .venv
 env/
 venv/
+venv1/
 ENV/
 env.bak/
 venv.bak/
@@ -133,6 +134,8 @@ dmypy.json
 
 # untracked dev files
 test-*.py
+test.txt
+topsecret.mkv
 
 # intelliJ’s project specific settings files
-.idea
+.idea

CHANGELOG.md

@@ -1,6 +1,6 @@
 # Changelog
 
-## Version 0.2.7 (2021-8-31)
+## Version 0.2.7 (2021-9-12)
 
 Makes further improvements to the CLI:
 
@@ -9,8 +9,11 @@ Makes further improvements to the CLI:
   with the `--no-check` method. It's enabled by default and may require further
   user-input in case the target directory contains a file with the same name as
   the issued download command
+- further adds a `--user-agent` and `--proxies` flag to the CLI; these fields were also
+  added to the `AnonFile` constructor
 - implements a preview command to obtain meta data without committing to a time-
   consuming download
+- commands related to the log file were also added to the CLI
 
 As for the main library, the following changes have been added since the last release:
 

README.md

@@ -84,8 +84,8 @@ the `AnonFile` API visit [anonfiles.com](https://anonfiles.com/docs/api).
 ## Command Line Interface
 
 ```bash
-# get help
-anonfile [download|upload] --help
+# open help page for specific commands
+anonfile [download|upload|preview|log] --help
 
 # note: both methods expect at least one argument, but can take on more
 anonfile download --url https://anonfiles.com/93k5x1ucu0/test_txt

src/anonfile/__init__.py

@@ -4,6 +4,7 @@
 
 import argparse
 import sys
+from collections import namedtuple
 from distutils.util import strtobool
 from pathlib import Path
 from typing import List
@@ -12,22 +13,32 @@
 from .anonfile import __version__
 
 
-def __print_dict(dictionary: dict, indent: int=4) -> None:
-    print("{\n%s\n}" % '\n'.join([f"\033[36m{indent*' '}{key}\033[0m: \033[32m{value}\033[0m" for key, value in dictionary.items()]))
+def __print_dict(dictionary: dict, indent: int = 4) -> None:
+    print("{\n%s\n}" % '\n'.join([f"\033[36m{indent * ' '}{key}\033[0m: \033[32m{value}\033[0m" for key, value in dictionary.items()]))
+
 
 def __from_file(path: Path) -> List[str]:
     with open(path, mode='r', encoding='utf-8') as file_handler:
         return [line.rstrip() for line in file_handler.readlines() if line[0] != '#']
 
+
+def format_proxies(proxies: str) -> dict:
+    return {prot: f"{prot}://{ip}" for (prot, ip) in [proxy.split('://') for proxy in proxies.split()]}
+
+
 def main():
     parser = argparse.ArgumentParser(prog=package_name)
     parser._positionals.title = 'Commands'
     parser._optionals.title = 'Arguments'
 
     parser.add_argument('-v', '--version', action='version', version=f"%(prog)s {__version__}")
-    parser.add_argument('-V', '--verbose', default=True, action=argparse.BooleanOptionalAction, help="increase output verbosity")
-    parser.add_argument('-l', '--logging', default=True, action=argparse.BooleanOptionalAction, help="enable URL logging")
+    parser.add_argument('-V', '--verbose', default=True, action='store_true', help="increase output verbosity (default)")
+    parser.add_argument('--no-verbose', dest='verbose', action='store_false', help="run commands silently")
+    parser.add_argument('-l', '--logging', default=True, action='store_true', help="enable URL logging (default)")
+    parser.add_argument('--no-logging', dest='logging', action='store_false', help="disable all logging activities")
     parser.add_argument('-t', '--token', type=str, default='secret', help="configure an API token (optional)")
+    parser.add_argument('-a', '--user-agent', type=str, default=None, help="configure custom User-Agent (optional)")
+    parser.add_argument('-p', '--proxies', type=str, default=None, help="configure HTTP and/or HTTPS proxies (optional)")
 
     subparser = parser.add_subparsers(dest='command')
     upload_parser = subparser.add_parser('upload', help="upload a file to https://anonfiles.com")
@@ -40,15 +51,24 @@ def main():
     download_parser.add_argument('-u', '--url', nargs='*', type=str, help="one or more URLs to download")
     download_parser.add_argument('-f', '--batch-file', type=Path, nargs='?', help="file containing URLs to download, one URL per line")
     download_parser.add_argument('-p', '--path', type=Path, default=Path.cwd(), help="download directory (CWD by default)")
-    download_parser.add_argument('-c', '--check', default=True, action=argparse.BooleanOptionalAction, help="check for duplicates")
+    download_parser.add_argument('-c', '--check', default=True, action='store_true', help="check for duplicates (default)")
+    download_parser.add_argument('--no-check', dest='check', action='store_false', help="disable checking for duplicates")
+
+    log_parser = subparser.add_parser('log', help="access the anonfile logger")
+    log_parser.add_argument('--reset', action='store_true', help="reset all log file entries")
+    log_parser.add_argument('--path', action='store_true', help="return the log file path")
+    log_parser.add_argument('--read', action='store_true', help='read the log file')
 
     try:
         args = parser.parse_args()
-        anon = AnonFile(args.token)
+        anon = AnonFile(args.token, user_agent=args.user_agent, proxies=format_proxies(args.proxies) if args.proxies else None)
 
         if args.command is None:
             raise UserWarning("missing a command")
 
+        if args.user_agent is not None:
+            anon.user_agent = args.user_agent
+
         if args.command == 'upload':
             for file in args.file:
                 upload = anon.upload(file, progressbar=args.verbose, enable_logging=args.logging)
@@ -57,7 +77,7 @@ def main():
         if args.command == 'preview':
             for url in args.url:
                 preview = anon.preview(url)
-                values = ['online' if preview.status else 'offline', preview.file_path.name, preview.url.geturl(), preview.ddl, preview.id, f"{preview.size}B"]
+                values = ['online' if preview.status else 'offline', preview.file_path.name, preview.url.geturl(), preview.ddl.geturl(), preview.id, f"{preview.size}B"]
 
                 if args.verbose:
                     __print_dict(dict(zip(['Status', 'File Path', 'URL', 'DDL', 'ID', 'Size'], values)))
@@ -76,6 +96,30 @@ def main():
                 else:
                     print(f"File: {download(url).file_path}")
 
+        if args.command == 'log':
+            if args.reset:
+                open(get_logfile_path(), mode='w', encoding='utf-8').close()
+            if args.path:
+                print(get_logfile_path())
+            if args.read:
+                with open(get_logfile_path(), mode='r', encoding='utf-8') as file_handler:
+                    log = file_handler.readlines()
+
+                    if not log:
+                        msg = "Nothing to read because the log file is empty"
+                        print(f"\033[33m{'[ WARNING ]'.ljust(12, ' ')}\033[0m{msg}")
+                        return
+
+                    parse = lambda line: line.strip('\n').split('::')
+                    Entry = namedtuple('Entry', 'timestamp method url')
+
+                    tabulate = "{:<19} {:<8} {:<30}".format
+
+                    print(f"\033[32m{tabulate('Date', 'Method', 'URL')}\033[0m")
+
+                    for line in log:
+                        entry = Entry(parse(line)[0], parse(line)[1], parse(line)[2])
+                        print(tabulate(entry.timestamp, entry.method, entry.url))
 
     except UserWarning as bad_human:
         print(f"error: {bad_human}")

src/anonfile/anonfile.py

@@ -34,7 +34,7 @@
 import sys
 from dataclasses import dataclass
 from pathlib import Path
-from typing import List, Tuple
+from typing import List, Tuple, Union
 from urllib.parse import ParseResult, urljoin, urlparse
 from urllib.request import getproxies
 
@@ -89,9 +89,13 @@ def get_logfile_path() -> Path:
 
 @dataclass(frozen=True)
 class ParseResponse:
+    """
+    Data class that is primarily used as a structured return type for the upload,
+    preview and download methods.
+    """
     response: Response
     file_path: Path
-    ddl: str
+    ddl: ParseResult
 
     @property
     def json(self) -> dict:
@@ -103,7 +107,7 @@ def json(self) -> dict:
     @property
     def status(self) -> bool:
         """
-        Return the upload status. If `false`, an error message indicating the
+        Return the upload status. If `False`, an error message indicating the
         cause for the malfunction will be redirected to `sys.stderr`.
         """
         status = bool(self.json['status'])
@@ -146,32 +150,61 @@ def size(self) -> int:
         """
         return int(self.json['data']['file']['metadata']['size']['bytes'])
 
+    def __str__(self) -> str:
+        return str(self.name)
+
+    def __repr__(self) -> str:
+        return f"{self.__class__.__name__}(ID={self.id})"
+
     #endregion
 
 class AnonFile:
+    """
+    The unofficial Python API for https://anonfiles.com.
+
+    Basic Usage
+    -----------
+
+    ```
+    from anonfile import AnonFile
+
+    anon = AnonFile()
+    preview = anon.preview('https://anonfiles.com/b7NaVd0cu3/topsecret_mkv')
+
+    # topsecret.mkv
+    print(preview)
+    ```
+
+    Docs
+    ----
+    See full documentation at <https://www.hentai-chan.dev/projects/anonfile>.
+    """
     _timeout = (5, 5)
     _total = 5
     _status_forcelist = [413, 429, 500, 502, 503, 504]
     _backoff_factor = 1
     _user_agent = None
+    _proxies = None
 
     API = "https://api.anonfiles.com/"
 
-    __slots__ = ['endpoint', 'token', 'timeout', 'total', 'status_forcelist', 'backoff_factor', 'user_agent']
+    __slots__ = ['endpoint', 'token', 'timeout', 'total', 'status_forcelist', 'backoff_factor', 'user_agent', 'proxies']
 
     def __init__(self,
                  token: str="undefined",
                  timeout: Tuple[float,float]=_timeout,
                  total: int=_total,
                  status_forcelist: List[int]=_status_forcelist,
                  backoff_factor: int=_backoff_factor,
-                 user_agent: str=_user_agent) -> AnonFile:
+                 user_agent: str=_user_agent,
+                 proxies: dict=_proxies) -> AnonFile:
         self.token = token
         self.timeout = timeout
         self.total = total,
         self.status_forcelist = status_forcelist,
         self.backoff_factor = backoff_factor
         self.user_agent = user_agent
+        self.proxies = proxies
 
     @staticmethod
     def __progressbar_options(iterable, desc, unit, color: str="\033[32m", char='\u25CB', total=None, disable=False) -> dict:
@@ -219,7 +252,7 @@ def __get(self, url: str, **kwargs) -> Response:
         Returns the GET request encoded in `utf-8`. Adds proxies to this session
         on the fly if urllib is able to pick up the system's proxy settings.
         """
-        response = self.session.get(url, timeout=self.timeout, proxies=getproxies(), **kwargs)
+        response = self.session.get(url, timeout=self.timeout, proxies=self.proxies or getproxies(), **kwargs)
         response.encoding = 'utf-8'
         return response
 
@@ -231,7 +264,7 @@ def __callback(monitor: MultipartEncoderMonitor, tqdm_handler: tqdm):
         tqdm_handler.total = monitor.len
         tqdm_handler.update(monitor.bytes_read - tqdm_handler.n)
 
-    def upload(self, path: str, progressbar: bool=False, enable_logging: bool=False) -> ParseResponse:
+    def upload(self, path: Union[str, Path], progressbar: bool=False, enable_logging: bool=False) -> ParseResponse:
         """
         Upload a file located in `path` to http://anonfiles.com. Set
         `enable_logging` to `True` to store the URL in a global config file.
@@ -252,13 +285,14 @@ def upload(self, path: str, progressbar: bool=False, enable_logging: bool=False)
         Note
         ----
         Although `anonfile` offers unlimited bandwidth, uploads cannot exceed a
-        file size of 20GB in theory. Due to technical difficulties in the implementation
+        file size of 20GB in theory. Due to technical difficulties in the implementation,
         the upper cap occurs much earlier at around 500MB.
         """
+        path = Path(path)
         size = os.stat(path).st_size
-        options = AnonFile.__progressbar_options(None, f"Upload: {Path(path).name}", unit='B', total=size, disable=progressbar)
+        options = AnonFile.__progressbar_options(None, f"Upload: {path.name}", unit='B', total=size, disable=progressbar)
         with open(path, mode='rb') as file_handler:
-            fields = {'file': (Path(path).name, file_handler, 'application/octet-stream')}
+            fields = {'file': (path.name, file_handler, 'application/octet-stream')}
             with tqdm(**options) as tqdm_handler:
                 encoder_monitor = MultipartEncoderMonitor.from_fields(fields, callback=lambda monitor: AnonFile.__callback(monitor, tqdm_handler))
                 response = self.session.post(
@@ -271,9 +305,9 @@ def upload(self, path: str, progressbar: bool=False, enable_logging: bool=False)
                     verify=True
                 )
                 logger.log(logging.INFO if enable_logging else logging.NOTSET, "upload::%s", response.json()['data']['file']['url']['full'])
-                return ParseResponse(response, Path(path), None)
+                return ParseResponse(response, path, None)
 
-    def preview(self, url: str, path: Path=Path.cwd()) -> ParseResponse:
+    def preview(self, url: str, path: Union[str, Path]=Path.cwd()) -> ParseResponse:
         """
         Obtain meta data associated with this `url` without commiting to a time-
         consuming download.
@@ -293,11 +327,11 @@ def preview(self, url: str, path: Path=Path.cwd()) -> ParseResponse:
         """
         with self.__get(urljoin(AnonFile.API, f"v2/file/{urlparse(url).path.split('/')[1]}/info")) as response:
             links = re.findall(r'''.*?href=['"](.*?)['"].*?''', html.unescape(self.__get(url).text), re.I)
-            ddl = next(filter(lambda link: 'cdn-' in link, links))
-            file_path = path.joinpath(Path(urlparse(ddl).path).name)
+            ddl = urlparse(next(filter(lambda link: 'cdn-' in link, links)))
+            file_path = Path(path).joinpath(Path(ddl.path).name)
             return ParseResponse(response, file_path, ddl)
 
-    def download(self, url: str, path: Path=Path.cwd(), progressbar: bool=False, enable_logging: bool=False) -> ParseResponse:
+    def download(self, url: str, path: Union[str, Path]=Path.cwd(), progressbar: bool=False, enable_logging: bool=False) -> ParseResponse:
         """
         Download a file from https://anonfiles.com given a `url`. Set the download
         directory in `path` (uses the current working directory by default). Set
@@ -328,7 +362,7 @@ def download(self, url: str, path: Path=Path.cwd(), progressbar: bool=False, ena
         options = AnonFile.__progressbar_options(None, f"Download {download.id}", unit='B', total=download.size, disable=progressbar)
         with open(download.file_path, mode='wb') as file_handler:
             with tqdm(**options) as tqdm_handler:
-                with self.__get(download.ddl, stream=True) as response:
+                with self.__get(download.ddl.geturl(), stream=True) as response:
                     for chunk in response.iter_content(1024*1024):
                         tqdm_handler.update(len(chunk))
                         file_handler.write(chunk)

tests/conftest.py

@@ -8,6 +8,7 @@ def pytest_addoption(parser):
         help="Secret anonfiles.com API token."
     )
 
+
 def pytest_generate_tests(metafunc):
     if 'token' in metafunc.fixturenames:
         metafunc.parametrize('token', metafunc.config.getoption('token'))