# Project export: WatchDog AI This document was generated by HackStack to give an AI agent context about a hackathon project. Sections are labeled with their provenance; content marked as truncated was cut to keep this document small. ## Project metadata - Hackathon: TreeHacks 2025 - Tagline: Anti-Rug AI agent for memecoins by frontrunning + sandwich attacks. Longer version: - Memecoin risk evaluator - Auto trade bot - Stops hard and soft rug by removing financial incentives - Devpost: https://devpost.com/software/watchdog-ai - GitHub: https://github.com/BdNWoG/watchdog-ai-final - Video: https://www.youtube.com/embed/gup3GO98B5s?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent - Result: winner (Taisu Ventures: Most Promising Web3 Project (4x Deeper Network Air)) - Team: 1 GitHub contributor(s) — BdNWoG (4 commits) ## Devpost submission (written by the team) ### Inspiration Crypto AI - especially AI agents - has been a hot topic within the industry with many players trying to identify new ways to incorporate AI agents into their builds. However, it seems that the most promising fits are still within that of memecoins - another hot topic this cycle. Memecoins are fun. But rugs are not. Yet, it seems like there is no way to stop rugs from happening since it is financially advantageous to do so... or is there? ps another inspiration is the web3 track "prompt" to use AI agents + AVS to do something crazy So why is this crazy. Because this essentially is a way for a person to do good, make money and stop memecoin rugs at the same time. You get to make money of the rugger (be it hard rug - liquidity removal or soft rug - sell offs). In this project, we align fun, profitability and security. ### What it does WatchDog AI in its essence is a memecoin anti-rugger, solving the problem by removing the financial incentives. The AI agent is designed to sit on top of a potential AVS, obtaining information from 1) DEX to analyse the riskiness and the possibility of rugs, 2) mempools for transactions and 3) operates as an MEV trading bot. 1) Memecoin evaluator The AI agent itself possesses the ability to evaluate the potential of future rugs based on past transactions and other data about the memecoin. It provides an evaluation on a risk scale. If the memecoin is deemed to be "risky" WatchDog AI would start listening to the mempool for potential malicious rugs. 2) Mempool listener WatchDog taps into the mempools of coins to identify suspicious transactions in real time. It could be able to detect rug attempts and react instantly by sacrificing gas fees to frontrun the rug attempt. 3) MEV attack Autotrader Identifying the rug attempt (for instance a large sell off), the agent could decide to frontrun the transaction with a large short and buy back after the the rug attempt has taken place. (Or even worse, find a way to obtain a flash loan to create a further leverage on the trade. ) This would greatly reduce the monetary gains of rugger. Else if the rugger attempts to remove liquidity, the bot could remove liquidity ahead of the rugger such that there are insufficient liquidity for the rugger to remove resulting in their loss before adding liquidity back into the pool. Technicals To implement this project the following were developped, A simulated DEX, A risk evaluating AI agent A mempool listener A frontrunning + sandwich attack trading bot. The latter few kind of combines to become a simulated AVS/can be easily moved on top of an AVS. The following demo will be conducted on a simulated DEX (since repeatedly rug testing a real dex would probably not be wise). As such, the technicals of the price mechanisms and all (simple) transaction modes were built out from scratch together with the associated mempool (bootstrap version). The AI agent is able to analyse the coin based on robust data and autonomously make trading decisions to profit and stop rugs. This part is the more fun part and probably the part that really steps out of current boundaries. For the functional demo refer to the media attached. Potential Future Steps This project framework can be deployed in an on-chain environment to become a fully verifiable trustless AI. A potential framework could be DEX -> dexscreener, AI agent -> WatchDog, AVS -> Eigen AVS, transactions -> Flashbots Closing/What can WatchDog potentially Do WatchDog AI is a real-time, trust-minimized DeFi security solution that uses AI agents detect malicious transactions. It directly addresses the 40%+ of crypto scams that come from rug pulls and malicious contract actions by preemptively intercepting them in the mempool. By leveraging frontrun and sandwich attacks, it proactively rescues user funds instead of merely labeling risky tokens. Unlike profit-driven MEV bots, WatchDog AI focuses on protecting users and recovering stolen funds, thereby establishing a new standard for real-time, trust-minimized security and greatly reducing rug pulls and malicious contract scams in DeFi—making rug pulls a thing of the past. ## README (from the GitHub repository) No README available. ## Detected evidence (automated analysis) Indexed codebase: 1143 recognized source files, 13499 KB. - C (language) — detected in the code - CSS (language) — detected in the code - HTML (language) — detected in the code - JavaScript (language) — detected in the code - Python (language) — detected in the code - Solidity (language) — claimed on Devpost, not found in the code - TypeScript (language) — claimed on Devpost, not found in the code ## Codebase structure (from repository index) ### Files (120 of 1455) ``` .gitignore avs/avs.py avs/templates/index.html frontrunner/frontrunner.py frontrunner/templates/index.html mempool/mempool.py mempool/templates/index.html myenv/bin/activate myenv/bin/activate.csh myenv/bin/activate.fish myenv/bin/Activate.ps1 myenv/bin/distro myenv/bin/dotenv myenv/bin/flask myenv/bin/httpx myenv/bin/normalizer myenv/bin/openai myenv/bin/pip myenv/bin/pip3 myenv/bin/pip3.12 myenv/bin/python myenv/bin/python3 myenv/bin/python3.12 myenv/bin/tqdm myenv/lib/python3.12/site-packages/aiohappyeyeballs-2.4.6.dist-info/INSTALLER myenv/lib/python3.12/site-packages/aiohappyeyeballs-2.4.6.dist-info/LICENSE myenv/lib/python3.12/site-packages/aiohappyeyeballs-2.4.6.dist-info/METADATA myenv/lib/python3.12/site-packages/aiohappyeyeballs-2.4.6.dist-info/RECORD myenv/lib/python3.12/site-packages/aiohappyeyeballs-2.4.6.dist-info/WHEEL myenv/lib/python3.12/site-packages/aiohappyeyeballs/__init__.py myenv/lib/python3.12/site-packages/aiohappyeyeballs/_staggered.py myenv/lib/python3.12/site-packages/aiohappyeyeballs/impl.py myenv/lib/python3.12/site-packages/aiohappyeyeballs/py.typed myenv/lib/python3.12/site-packages/aiohappyeyeballs/types.py myenv/lib/python3.12/site-packages/aiohappyeyeballs/utils.py myenv/lib/python3.12/site-packages/aiohttp-3.11.12.dist-info/INSTALLER myenv/lib/python3.12/site-packages/aiohttp-3.11.12.dist-info/LICENSE.txt myenv/lib/python3.12/site-packages/aiohttp-3.11.12.dist-info/METADATA myenv/lib/python3.12/site-packages/aiohttp-3.11.12.dist-info/RECORD myenv/lib/python3.12/site-packages/aiohttp-3.11.12.dist-info/top_level.txt myenv/lib/python3.12/site-packages/aiohttp-3.11.12.dist-info/WHEEL myenv/lib/python3.12/site-packages/aiohttp/__init__.py myenv/lib/python3.12/site-packages/aiohttp/_cparser.pxd myenv/lib/python3.12/site-packages/aiohttp/_find_header.pxd myenv/lib/python3.12/site-packages/aiohttp/_headers.pxi myenv/lib/python3.12/site-packages/aiohttp/_http_parser.pyx myenv/lib/python3.12/site-packages/aiohttp/_http_writer.pyx myenv/lib/python3.12/site-packages/aiohttp/_websocket/__init__.py myenv/lib/python3.12/site-packages/aiohttp/_websocket/.hash/mask.pxd.hash myenv/lib/python3.12/site-packages/aiohttp/_websocket/.hash/mask.pyx.hash myenv/lib/python3.12/site-packages/aiohttp/_websocket/.hash/reader_c.pxd.hash myenv/lib/python3.12/site-packages/aiohttp/_websocket/helpers.py myenv/lib/python3.12/site-packages/aiohttp/_websocket/mask.pxd myenv/lib/python3.12/site-packages/aiohttp/_websocket/mask.pyx myenv/lib/python3.12/site-packages/aiohttp/_websocket/models.py myenv/lib/python3.12/site-packages/aiohttp/_websocket/reader_c.pxd myenv/lib/python3.12/site-packages/aiohttp/_websocket/reader_c.py myenv/lib/python3.12/site-packages/aiohttp/_websocket/reader_py.py myenv/lib/python3.12/site-packages/aiohttp/_websocket/reader.py myenv/lib/python3.12/site-packages/aiohttp/_websocket/writer.py myenv/lib/python3.12/site-packages/aiohttp/.hash/_cparser.pxd.hash myenv/lib/python3.12/site-packages/aiohttp/.hash/_find_header.pxd.hash myenv/lib/python3.12/site-packages/aiohttp/.hash/_http_parser.pyx.hash myenv/lib/python3.12/site-packages/aiohttp/.hash/_http_writer.pyx.hash myenv/lib/python3.12/site-packages/aiohttp/.hash/hdrs.py.hash myenv/lib/python3.12/site-packages/aiohttp/abc.py myenv/lib/python3.12/site-packages/aiohttp/base_protocol.py myenv/lib/python3.12/site-packages/aiohttp/client_exceptions.py myenv/lib/python3.12/site-packages/aiohttp/client_proto.py myenv/lib/python3.12/site-packages/aiohttp/client_reqrep.py myenv/lib/python3.12/site-packages/aiohttp/client_ws.py myenv/lib/python3.12/site-packages/aiohttp/client.py myenv/lib/python3.12/site-packages/aiohttp/compression_utils.py myenv/lib/python3.12/site-packages/aiohttp/connector.py myenv/lib/python3.12/site-packages/aiohttp/cookiejar.py myenv/lib/python3.12/site-packages/aiohttp/formdata.py myenv/lib/python3.12/site-packages/aiohttp/hdrs.py myenv/lib/python3.12/site-packages/aiohttp/helpers.py myenv/lib/python3.12/site-packages/aiohttp/http_exceptions.py myenv/lib/python3.12/site-packages/aiohttp/http_parser.py myenv/lib/python3.12/site-packages/aiohttp/http_websocket.py myenv/lib/python3.12/site-packages/aiohttp/http_writer.py myenv/lib/python3.12/site-packages/aiohttp/http.py myenv/lib/python3.12/site-packages/aiohttp/log.py myenv/lib/python3.12/site-packages/aiohttp/multipart.py myenv/lib/python3.12/site-packages/aiohttp/payload_streamer.py myenv/lib/python3.12/site-packages/aiohttp/payload.py myenv/lib/python3.12/site-packages/aiohttp/py.typed myenv/lib/python3.12/site-packages/aiohttp/pytest_plugin.py myenv/lib/python3.12/site-packages/aiohttp/resolver.py myenv/lib/python3.12/site-packages/aiohttp/streams.py myenv/lib/python3.12/site-packages/aiohttp/tcp_helpers.py myenv/lib/python3.12/site-packages/aiohttp/test_utils.py myenv/lib/python3.12/site-packages/aiohttp/tracing.py myenv/lib/python3.12/site-packages/aiohttp/typedefs.py myenv/lib/python3.12/site-packages/aiohttp/web_app.py myenv/lib/python3.12/site-packages/aiohttp/web_exceptions.py myenv/lib/python3.12/site-packages/aiohttp/web_fileresponse.py myenv/lib/python3.12/site-packages/aiohttp/web_log.py myenv/lib/python3.12/site-packages/aiohttp/web_middlewares.py myenv/lib/python3.12/site-packages/aiohttp/web_protocol.py myenv/lib/python3.12/site-packages/aiohttp/web_request.py myenv/lib/python3.12/site-packages/aiohttp/web_response.py myenv/lib/python3.12/site-packages/aiohttp/web_routedef.py myenv/lib/python3.12/site-packages/aiohttp/web_runner.py myenv/lib/python3.12/site-packages/aiohttp/web_server.py myenv/lib/python3.12/site-packages/aiohttp/web_urldispatcher.py myenv/lib/python3.12/site-packages/aiohttp/web_ws.py myenv/lib/python3.12/site-packages/aiohttp/web.py myenv/lib/python3.12/site-packages/aiohttp/worker.py myenv/lib/python3.12/site-packages/aiosignal-1.3.2.dist-info/INSTALLER myenv/lib/python3.12/site-packages/aiosignal-1.3.2.dist-info/LICENSE myenv/lib/python3.12/site-packages/aiosignal-1.3.2.dist-info/METADATA myenv/lib/python3.12/site-packages/aiosignal-1.3.2.dist-info/RECORD myenv/lib/python3.12/site-packages/aiosignal-1.3.2.dist-info/top_level.txt myenv/lib/python3.12/site-packages/aiosignal-1.3.2.dist-info/WHEEL myenv/lib/python3.12/site-packages/aiosignal/__init__.py myenv/lib/python3.12/site-packages/aiosignal/__init__.pyi myenv/lib/python3.12/site-packages/aiosignal/py.typed myenv/lib/python3.12/site-packages/annotated_types-0.7.0.dist-info/INSTALLER [1335 more files omitted for size] ``` ### Dependencies No dependency index available. ### Recent commits (newest first) - save - UI - change - first commit ## Key source files (fetched from GitHub, selected and truncated for size) ### myenv/lib/python3.12/site-packages/idna-3.10.dist-info/LICENSE.md ```markdown BSD 3-Clause License Copyright (c) 2013-2024, Kim Davies and contributors. All rights reserved. Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met: 1. Redistributions of source code must retain the above copyright notice, this list of conditions and the following disclaimer. 2. Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the following disclaimer in the documentation and/or other materials provided with the distribution. 3. Neither the name of the copyright holder nor the names of its contributors may be used to endorse or promote products derived from this software without specific prior written permission. THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. ``` ### myenv/lib/python3.12/site-packages/httpx-0.28.1.dist-info/licenses/LICENSE.md ```markdown Copyright © 2019, [Encode OSS Ltd](https://www.encode.io/). All rights reserved. Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met: * Redistributions of source code must retain the above copyright notice, this list of conditions and the following disclaimer. * Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the following disclaimer in the documentation and/or other materials provided with the distribution. * Neither the name of the copyright holder nor the names of its contributors may be used to endorse or promote products derived from this software without specific prior written permission. THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. ``` ### myenv/lib/python3.12/site-packages/dotenv/cli.py ```python import json import os import shlex import sys from contextlib import contextmanager from subprocess import Popen from typing import Any, Dict, IO, Iterator, List try: import click except ImportError: sys.stderr.write('It seems python-dotenv is not installed with cli option. \n' 'Run pip install "python-dotenv[cli]" to fix this.') sys.exit(1) from .main import dotenv_values, set_key, unset_key from .version import __version__ def enumerate_env(): """ Return a path for the ${pwd}/.env file. If pwd does not exist, return None. """ try: cwd = os.getcwd() except FileNotFoundError: return None path = os.path.join(cwd, '.env') return path @click.group() @click.option('-f', '--file', default=enumerate_env(), type=click.Path(file_okay=True), help="Location of the .env file, defaults to .env file in current working directory.") @click.option('-q', '--quote', default='always', type=click.Choice(['always', 'never', 'auto']), help="Whether to quote or not the variable values. Default mode is always. This does not affect parsing.") @click.option('-e', '--export', default=False, type=click.BOOL, help="Whether to write the dot file as an executable bash script.") @click.version_option(version=__version__) @click.pass_context def cli(ctx: click.Context, file: Any, quote: Any, export: Any) -> None: """This script is used to set, get or unset values from a .env file.""" ctx.obj = {'QUOTE': quote, 'EXPORT': export, 'FILE': file} @contextmanager def stream_file(path: os.PathLike) -> Iterator[IO[str]]: """ Open a file and yield the corresponding (decoded) stream. Exits with error code 2 if the file cannot be opened. """ try: with open(path) as stream: yield stream except OSError as exc: print(f"Error opening env file: {exc}", file=sys.stderr) exit(2) @cli.command() @click.pass_context @click.option('--format', default='simple', type=click.Choice(['simple', 'json', 'shell', 'export']), help="The format in which to display the list. Default format is simple, " "which displays name=value without quotes.") def list(ctx: click.Context, format: bool) -> None: """Display all the stored key/value.""" file = ctx.obj['FILE'] with stream_file(file) as stream: values = dotenv_values(stream=stream) if format == 'json': click.echo(json.dumps(values, indent=2, sort_keys=True)) else: prefix = 'export ' if format == 'export' else '' for k in sorted(values): v = values[k] if v is not None: if format in ('export', 'shell'): v = shlex.quote(v) click.echo(f'{prefix}{k}={v}') @cli.command() @click.pass_context @click.argument('key', required=True) @click.argument('value', required=True) def set(ctx: click.Context, key: Any, value: Any) -> None: """Store the given key/value.""" file = ctx.obj['FILE'] quote = ctx.obj['QUOTE'] export = ctx.obj['EXPORT'] success, key, value = set_key(file, key, value, quote, export) if success: click.echo(f'{key}={value}') else: exit(1) @cli.command() @click.pass_context @click.argument('key', required=True) def get(ctx: click.Context, key: Any) -> None: """Retrieve the value for the given key.""" file = ctx.obj['FILE'] with stream_file(file) as stream: values = dotenv_values(stream=stream) stored_value = values.get(key) if stored_value: click.echo(stored_value) else: exit(1) @cli.command() @click.pass_context @click.argument('key', required=True) def unset(ctx: click.Context, key: Any) -> None: """Removes the given key.""" file = ctx.obj['FILE'] quote = ctx.obj['QUOTE'] success, key = unset_key(file, key, quote) if success: click.echo(f"Successfully removed {key}") else: exit(1) @cli.command(context_settings={'ignore_unknown_options': True}) @click.pass_context @click.option( "--override/--no-override", default=True, help="Override variables from the environment file with those from the .env file.", ) @click.argument('commandline', nargs=-1, type=click.UNPROCESSED) def run(ctx: click.Context, override: bool, commandline: List[str]) -> None: """Run command with environment variables present.""" file = ctx.obj['FILE'] if not os.path.isfile(file): raise click.BadParameter( f'Invalid value for \'-f\' "{file}" does not exist.', ctx=ctx ) dotenv_as_dict = { k: v for (k, v) in dotenv_values(file).items() if v is not None and (override or k not in os.environ) } if not commandline: click.echo('No command given.') exit(1) ret = run_command(commandline, dotenv_as_dict) exit(ret) def run_command(command: List[str], env: Dict[str, str]) -> int: """Run command in sub process. Runs the command in a sub process with the variables from `env` added in the current environment variables. Parameters ---------- command: List[str] The command and it's parameters env: Dict The additional environment variables Returns ------- int The return code of the command """ # copy the current environment variables and add the vales from # `env` cmd_env = os.environ.copy() cmd_env.update(env) p = Popen(command, universal_newlines=True, bufsize=0, shell=False, env=cmd_env) _, _ = p.communicate() return p.returncode ``` ### myenv/lib/python3.12/site-packages/tqdm/cli.py ```python """ Module version for monitoring CLI pipes (`... | python -m tqdm | ...`). """ import logging import re import sys from ast import literal_eval as numeric from textwrap import indent from .std import TqdmKeyError, TqdmTypeError, tqdm from .version import __version__ __all__ = ["main"] log = logging.getLogger(__name__) def cast(val, typ): log.debug((val, typ)) if " or " in typ: for t in typ.split(" or "): try: return cast(val, t) except TqdmTypeError: pass raise TqdmTypeError(f"{val} : {typ}") # sys.stderr.write('\ndebug | `val:type`: `' + val + ':' + typ + '`.\n') if typ == 'bool': if (val == 'True') or (val == ''): return True if val == 'False': return False raise TqdmTypeError(val + ' : ' + typ) if typ == 'chr': if len(val) == 1: return val.encode() if re.match(r"^\\\w+$", val): return eval(f'"{val}"').encode() raise TqdmTypeError(f"{val} : {typ}") if typ == 'str': return val if typ == 'int': try: return int(val) except ValueError as exc: raise TqdmTypeError(f"{val} : {typ}") from exc if typ == 'float': try: return float(val) except ValueError as exc: raise TqdmTypeError(f"{val} : {typ}") from exc raise TqdmTypeError(f"{val} : {typ}") def posix_pipe(fin, fout, delim=b'\\n', buf_size=256, callback=lambda float: None, callback_len=True): """ Params ------ fin : binary file with `read(buf_size : int)` method fout : binary file with `write` (and optionally `flush`) methods. callback : function(float), e.g.: `tqdm.update` callback_len : If (default: True) do `callback(len(buffer))`. Otherwise, do `callback(data) for data in buffer.split(delim)`. """ fp_write = fout.write if not delim: while True: tmp = fin.read(buf_size) # flush at EOF if not tmp: getattr(fout, 'flush', lambda: None)() return fp_write(tmp) callback(len(tmp)) # return buf = b'' len_delim = len(delim) # n = 0 while True: tmp = fin.read(buf_size) # flush at EOF if not tmp: if buf: fp_write(buf) if callback_len: # n += 1 + buf.count(delim) callback(1 + buf.count(delim)) else: for i in buf.split(delim): callback(i) getattr(fout, 'flush', lambda: None)() return # n while True: i = tmp.find(delim) if i < 0: buf += tmp break fp_write(buf + tmp[:i + len(delim)]) # n += 1 callback(1 if callback_len else (buf + tmp[:i])) buf = b'' tmp = tmp[i + len_delim:] # ((opt, type), ... ) RE_OPTS = re.compile(r'\n {4}(\S+)\s{2,}:\s*([^,]+)') # better split method assuming no positional args RE_SHLEX = re.compile(r'\s*(? : \2', d) split = RE_OPTS.split( [truncated — 5010 more characters] ``` ### myenv/lib/python3.12/site-packages/dotenv/main.py ```python import io import logging import os import pathlib import shutil import sys import tempfile from collections import OrderedDict from contextlib import contextmanager from typing import (IO, Dict, Iterable, Iterator, Mapping, Optional, Tuple, Union) from .parser import Binding, parse_stream from .variables import parse_variables # A type alias for a string path to be used for the paths in this file. # These paths may flow to `open()` and `shutil.move()`; `shutil.move()` # only accepts string paths, not byte paths or file descriptors. See # https://github.com/python/typeshed/pull/6832. StrPath = Union[str, 'os.PathLike[str]'] logger = logging.getLogger(__name__) def with_warn_for_invalid_lines(mappings: Iterator[Binding]) -> Iterator[Binding]: for mapping in mappings: if mapping.error: logger.warning( "Python-dotenv could not parse statement starting at line %s", mapping.original.line, ) yield mapping class DotEnv: def __init__( self, dotenv_path: Optional[StrPath], stream: Optional[IO[str]] = None, verbose: bool = False, encoding: Optional[str] = None, interpolate: bool = True, override: bool = True, ) -> None: self.dotenv_path: Optional[StrPath] = dotenv_path self.stream: Optional[IO[str]] = stream self._dict: Optional[Dict[str, Optional[str]]] = None self.verbose: bool = verbose self.encoding: Optional[str] = encoding self.interpolate: bool = interpolate self.override: bool = override @contextmanager def _get_stream(self) -> Iterator[IO[str]]: if self.dotenv_path and os.path.isfile(self.dotenv_path): with open(self.dotenv_path, encoding=self.encoding) as stream: yield stream elif self.stream is not None: yield self.stream else: if self.verbose: logger.info( "Python-dotenv could not find configuration file %s.", self.dotenv_path or '.env', ) yield io.StringIO('') def dict(self) -> Dict[str, Optional[str]]: """Return dotenv as dict""" if self._dict: return self._dict raw_values = self.parse() if self.interpolate: self._dict = OrderedDict(resolve_variables(raw_values, override=self.override)) else: self._dict = OrderedDict(raw_values) return self._dict def parse(self) -> Iterator[Tuple[str, Optional[str]]]: with self._get_stream() as stream: for mapping in with_warn_for_invalid_lines(parse_stream(stream)): if mapping.key is not None: yield mapping.key, mapping.value def set_as_environment_variables(self) -> bool: """ Load the current dotenv as system environment variable. """ if not self.dict(): return False for k, v in self.dict().items(): if k in os.environ and not self.override: continue if v is not None: os.environ[k] = v return True def get(self, key: str) -> Optional[str]: """ """ data = self.dict() if key in data: return data[key] if self.verbose: logger.warning("Key %s not found in %s.", key, self.dotenv_path) return None def get_key( dotenv_path: StrPath, key_to_get: str, encoding: Optional[str] = "utf-8", ) -> Optional[str]: """ Get the value of a given key from the given .env. Returns `None` if the key isn't found or doesn't have a value. """ return DotEnv(dotenv_path, verbose=True, encoding=encoding).get(key_to_get) @contextmanager def rewrite( path: StrPath, encoding: Optional[str], ) -> Iterator[Tuple[IO[str], IO[str]]]: pathlib.Path(path).touch() with tempfile.NamedTemporaryFile(mode="w", encoding=encoding, delete=False) as dest: error = None try: with open(path, encoding=encoding) as source: yield (source, dest) except BaseException as err: error = err if error is None: shutil.move(dest.name, path) else: os.unlink(dest.name) raise error from None def set_key( dotenv_path: StrPath, key_to_set: str, value_to_set: str, quote_mode: str = "always", export: bool = False, encoding: Optional[str] = "utf-8", ) -> Tuple[Optional[bool], str, str]: """ Adds or Updates a key/value to the given .env If the .env path given doesn't exist, fails instead of risking creating an orphan .env somewhere in the filesystem """ if quote_mode not in ("always", "auto", "never"): raise ValueError(f"Unknown quote_mode: {quote_mode}") quote = ( quote_mode == "always" or (quote_mode == "auto" and not value_to_set.isalnum()) ) if quote: value_out = "'{}'".format(value_to_set.replace("'", "\\'")) else: value_out = value_to_set if export: line_out = f'export {key_to_set}={value_out}\n' else: line_out = f"{key_to_set}={value_out}\n" with rewrite(dotenv_path, encoding=encoding) as (source, dest): replaced = False missing_newline = False for mapping in with_warn_for_invalid_lines(parse_stream(source)): if mapping.key == key_to_set: dest.write(line_out) replaced = True else: dest.write(mapping.original.string) missing_newline = not mapping.original.string.endswith("\n") if not replaced: if missing_newline: dest.write("\n") dest.write(line_out) return True, key_to_set, value_to_set def unset_key( dotenv_path: StrPath, key_to_ [truncated — 6098 more characters] ``` ### myenv/lib/python3.12/site-packages/pip/_internal/main.py ```python from typing import List, Optional def main(args: Optional[List[str]] = None) -> int: """This is preserved for old console scripts that may still be referencing it. For additional details, see https://github.com/pypa/pip/issues/7498. """ from pip._internal.utils.entrypoints import _wrapper return _wrapper(args) ``` ### myenv/lib/python3.12/site-packages/flask/cli.py ```python from __future__ import annotations import ast import collections.abc as cabc import importlib.metadata import inspect import os import platform import re import sys import traceback import typing as t from functools import update_wrapper from operator import itemgetter from types import ModuleType import click from click.core import ParameterSource from werkzeug import run_simple from werkzeug.serving import is_running_from_reloader from werkzeug.utils import import_string from .globals import current_app from .helpers import get_debug_flag from .helpers import get_load_dotenv if t.TYPE_CHECKING: import ssl from _typeshed.wsgi import StartResponse from _typeshed.wsgi import WSGIApplication from _typeshed.wsgi import WSGIEnvironment from .app import Flask class NoAppException(click.UsageError): """Raised if an application cannot be found or loaded.""" def find_best_app(module: ModuleType) -> Flask: """Given a module instance this tries to find the best possible application in the module or raises an exception. """ from . import Flask # Search for the most common names first. for attr_name in ("app", "application"): app = getattr(module, attr_name, None) if isinstance(app, Flask): return app # Otherwise find the only object that is a Flask instance. matches = [v for v in module.__dict__.values() if isinstance(v, Flask)] if len(matches) == 1: return matches[0] elif len(matches) > 1: raise NoAppException( "Detected multiple Flask applications in module" f" '{module.__name__}'. Use '{module.__name__}:name'" " to specify the correct one." ) # Search for app factory functions. for attr_name in ("create_app", "make_app"): app_factory = getattr(module, attr_name, None) if inspect.isfunction(app_factory): try: app = app_factory() if isinstance(app, Flask): return app except TypeError as e: if not _called_with_wrong_args(app_factory): raise raise NoAppException( f"Detected factory '{attr_name}' in module '{module.__name__}'," " but could not call it without arguments. Use" f" '{module.__name__}:{attr_name}(args)'" " to specify arguments." ) from e raise NoAppException( "Failed to find Flask application or factory in module" f" '{module.__name__}'. Use '{module.__name__}:name'" " to specify one." ) def _called_with_wrong_args(f: t.Callable[..., Flask]) -> bool: """Check whether calling a function raised a ``TypeError`` because the call failed or because something in the factory raised the error. :param f: The function that was called. :return: ``True`` if the call failed. """ tb = sys.exc_info()[2] try: while tb is not None: if tb.tb_frame.f_code is f.__code__: # In the function, it was called successfully. return False tb = tb.tb_next # Didn't reach the function. return True finally: # Delete tb to break a circular reference. # https://docs.python.org/2/library/sys.html#sys.exc_info del tb def find_app_by_string(module: ModuleType, app_name: str) -> Flask: """Check if the given string is a variable name or a function. Call a function to get the app instance, or return the variable directly. """ from . import Flask # Parse app_name as a single expression to determine if it's a valid # attribute name or function call. try: expr = ast.parse(app_name.strip(), mode="eval").body except SyntaxError: raise NoAppException( f"Failed to parse {app_name!r} as an attribute name or function call." ) from None if isinstance(expr, ast.Name): name = expr.id args = [] kwargs = {} elif isinstance(expr, ast.Call): # Ensure the function name is an attribute name only. if not isinstance(expr.func, ast.Name): raise NoAppException( f"Function reference must be a simple name: {app_name!r}." ) name = expr.func.id # Parse the positional and keyword arguments as literals. try: args = [ast.literal_eval(arg) for arg in expr.args] kwargs = { kw.arg: ast.literal_eval(kw.value) for kw in expr.keywords if kw.arg is not None } except ValueError: # literal_eval gives cryptic error messages, show a generic # message with the full expression instead. raise NoAppException( f"Failed to parse arguments as literal values: {app_name!r}." ) from None else: raise NoAppException( f"Failed to parse {app_name!r} as an attribute name or function call." ) try: attr = getattr(module, name) except AttributeError as e: raise NoAppException( f"Failed to find attribute {name!r} in {module.__name__!r}." ) from e # If the attribute is a function, call it with any args and kwargs # to get the real application. if inspect.isfunction(attr): try: app = attr(*args, **kwargs) except TypeError as e: if not _called_with_wrong_args(attr): raise raise NoAppException( f"The factory {app_name!r} in module" f" {module.__name__!r} could not be called with the" " specified arguments." ) from e else: app = attr if isinstance(app, Flask): return app raise NoAppException( "A valid Flask application [truncated — 31093 more characters] ``` ### myenv/lib/python3.12/site-packages/pip/_internal/models/index.py ```python import urllib.parse class PackageIndex: """Represents a Package Index and provides easier access to endpoints""" __slots__ = ["url", "netloc", "simple_url", "pypi_url", "file_storage_domain"] def __init__(self, url: str, file_storage_domain: str) -> None: super().__init__() self.url = url self.netloc = urllib.parse.urlsplit(url).netloc self.simple_url = self._url_for_path("simple") self.pypi_url = self._url_for_path("pypi") # This is part of a temporary hack used to block installs of PyPI # packages which depend on external urls only necessary until PyPI can # block such packages themselves self.file_storage_domain = file_storage_domain def _url_for_path(self, path: str) -> str: return urllib.parse.urljoin(self.url, path) PyPI = PackageIndex("https://pypi.org/", file_storage_domain="files.pythonhosted.org") TestPyPI = PackageIndex( "https://test.pypi.org/", file_storage_domain="test-files.pythonhosted.org" ) ``` ### myenv/lib/python3.12/site-packages/pip/_internal/cli/main.py ```python """Primary application entrypoint. """ import locale import logging import os import sys import warnings from typing import List, Optional from pip._internal.cli.autocompletion import autocomplete from pip._internal.cli.main_parser import parse_command from pip._internal.commands import create_command from pip._internal.exceptions import PipError from pip._internal.utils import deprecation logger = logging.getLogger(__name__) # Do not import and use main() directly! Using it directly is actively # discouraged by pip's maintainers. The name, location and behavior of # this function is subject to change, so calling it directly is not # portable across different pip versions. # In addition, running pip in-process is unsupported and unsafe. This is # elaborated in detail at # https://pip.pypa.io/en/stable/user_guide/#using-pip-from-your-program. # That document also provides suggestions that should work for nearly # all users that are considering importing and using main() directly. # However, we know that certain users will still want to invoke pip # in-process. If you understand and accept the implications of using pip # in an unsupported manner, the best approach is to use runpy to avoid # depending on the exact location of this entry point. # The following example shows how to use runpy to invoke pip in that # case: # # sys.argv = ["pip", your, args, here] # runpy.run_module("pip", run_name="__main__") # # Note that this will exit the process after running, unlike a direct # call to main. As it is not safe to do any processing after calling # main, this should not be an issue in practice. def main(args: Optional[List[str]] = None) -> int: if args is None: args = sys.argv[1:] # Suppress the pkg_resources deprecation warning # Note - we use a module of .*pkg_resources to cover # the normal case (pip._vendor.pkg_resources) and the # devendored case (a bare pkg_resources) warnings.filterwarnings( action="ignore", category=DeprecationWarning, module=".*pkg_resources" ) # Configure our deprecation warnings to be sent through loggers deprecation.install_warning_logger() autocomplete() try: cmd_name, cmd_args = parse_command(args) except PipError as exc: sys.stderr.write(f"ERROR: {exc}") sys.stderr.write(os.linesep) sys.exit(1) # Needed for locale.getpreferredencoding(False) to work # in pip._internal.utils.encoding.auto_decode try: locale.setlocale(locale.LC_ALL, "") except locale.Error as e: # setlocale can apparently crash if locale are uninitialized logger.debug("Ignoring error %s when setting locale", e) command = create_command(cmd_name, isolated=("--isolated" in cmd_args)) return command.main(cmd_args) ``` ### myenv/lib/python3.12/site-packages/pip/_internal/commands/index.py ```python import logging from optparse import Values from typing import Any, Iterable, List, Optional, Union from pip._vendor.packaging.version import LegacyVersion, Version from pip._internal.cli import cmdoptions from pip._internal.cli.req_command import IndexGroupCommand from pip._internal.cli.status_codes import ERROR, SUCCESS from pip._internal.commands.search import print_dist_installation_info from pip._internal.exceptions import CommandError, DistributionNotFound, PipError from pip._internal.index.collector import LinkCollector from pip._internal.index.package_finder import PackageFinder from pip._internal.models.selection_prefs import SelectionPreferences from pip._internal.models.target_python import TargetPython from pip._internal.network.session import PipSession from pip._internal.utils.misc import write_output logger = logging.getLogger(__name__) class IndexCommand(IndexGroupCommand): """ Inspect information available from package indexes. """ ignore_require_venv = True usage = """ %prog versions """ def add_options(self) -> None: cmdoptions.add_target_python_options(self.cmd_opts) self.cmd_opts.add_option(cmdoptions.ignore_requires_python()) self.cmd_opts.add_option(cmdoptions.pre()) self.cmd_opts.add_option(cmdoptions.no_binary()) self.cmd_opts.add_option(cmdoptions.only_binary()) index_opts = cmdoptions.make_option_group( cmdoptions.index_group, self.parser, ) self.parser.insert_option_group(0, index_opts) self.parser.insert_option_group(0, self.cmd_opts) def run(self, options: Values, args: List[str]) -> int: handlers = { "versions": self.get_available_package_versions, } logger.warning( "pip index is currently an experimental command. " "It may be removed/changed in a future release " "without prior warning." ) # Determine action if not args or args[0] not in handlers: logger.error( "Need an action (%s) to perform.", ", ".join(sorted(handlers)), ) return ERROR action = args[0] # Error handling happens here, not in the action-handlers. try: handlers[action](options, args[1:]) except PipError as e: logger.error(e.args[0]) return ERROR return SUCCESS def _build_package_finder( self, options: Values, session: PipSession, target_python: Optional[TargetPython] = None, ignore_requires_python: Optional[bool] = None, ) -> PackageFinder: """ Create a package finder appropriate to the index command. """ link_collector = LinkCollector.create(session, options=options) # Pass allow_yanked=False to ignore yanked versions. selection_prefs = SelectionPreferences( allow_yanked=False, allow_all_prereleases=options.pre, ignore_requires_python=ignore_requires_python, ) return PackageFinder.create( link_collector=link_collector, selection_prefs=selection_prefs, target_python=target_python, ) def get_available_package_versions(self, options: Values, args: List[Any]) -> None: if len(args) != 1: raise CommandError("You need to specify exactly one argument") target_python = cmdoptions.make_target_python(options) query = args[0] with self._build_session(options) as session: finder = self._build_package_finder( options=options, session=session, target_python=target_python, ignore_requires_python=options.ignore_requires_python, ) versions: Iterable[Union[LegacyVersion, Version]] = ( candidate.version for candidate in finder.find_all_candidates(query) ) if not options.pre: # Remove prereleases versions = ( version for version in versions if not version.is_prerelease ) versions = set(versions) if not versions: raise DistributionNotFound( f"No matching distribution found for {query}" ) formatted_versions = [str(ver) for ver in sorted(versions, reverse=True)] latest = formatted_versions[0] write_output(f"{query} ({latest})") write_output("Available versions: {}".format(", ".join(formatted_versions))) print_dist_installation_info(query, latest) ``` [1060 more indexed source files omitted to keep this export small. The full file list is in the Codebase structure section above.]