We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Directories of exploded archives not recognized as cluster
Directories of exploded archives aren't recognized as a cluster but are erroneously identified as whatever context matches the first marker file we encounter after recursively enumerating every file in all subdirectories.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `insights/core/hydration.py`
Content:
```
1 import logging
2 import os
3 from itertools import product
4 
5 from insights.core import archives
6 from insights.core.context import (ClusterArchiveContext,
7                                    JDRContext,
8                                    HostArchiveContext,
9                                    SosArchiveContext,
10                                    SerializedArchiveContext)
11 
12 log = logging.getLogger(__name__)
13 
14 
15 def get_all_files(path):
16     all_files = []
17     for f in archives.get_all_files(path):
18         if os.path.isfile(f) and not os.path.islink(f):
19             all_files.append(f)
20     return all_files
21 
22 
23 def identify(files):
24     markers = {"insights_archive.txt": SerializedArchiveContext,
25                "insights_commands": HostArchiveContext,
26                "sos_commands": SosArchiveContext,
27                "JBOSS_HOME": JDRContext}
28 
29     for f, m in product(files, markers):
30         if m in f:
31             i = f.find(m)
32             common_path = os.path.dirname(f[:i])
33             ctx = markers[m]
34             return common_path, ctx
35 
36     common_path = os.path.dirname(os.path.commonprefix(files))
37     if not common_path:
38         raise archives.InvalidArchive("Unable to determine common path")
39 
40     if any(f.endswith(archives.COMPRESSION_TYPES) for f in os.listdir(common_path)):
41         return common_path, ClusterArchiveContext
42 
43     return common_path, HostArchiveContext
44 
45 
46 def create_context(path, context=None):
47     all_files = get_all_files(path)
48     if not all_files:
49         raise archives.InvalidArchive("No files in archive")
50 
51     common_path, ctx = identify(all_files)
52     context = context or ctx
53     return context(common_path, all_files=all_files)
54
```

Path: `insights/core/cluster.py`
Content:
```
1 #!/usr/bin/env python
2 import itertools
3 import pandas as pd
4 from collections import defaultdict
5 
6 from ansible.parsing.dataloader import DataLoader
7 from ansible.inventory.manager import InventoryManager
8 
9 from insights.core import dr, plugins
10 from insights.core.archives import extract
11 from insights.core.hydration import create_context
12 from insights.specs import Specs
13 
14 
15 ID_GENERATOR = itertools.count()
16 
17 
18 class ClusterMeta(dict):
19     def __init__(self, num_members, kwargs):
20         self.num_members = num_members
21         self.update(**kwargs)
22 
23 
24 @plugins.combiner(optional=[Specs.machine_id, Specs.hostname])
25 def machine_id(mid, hn):
26     ds = mid or hn
27     if ds:
28         return ds.content[0].strip()
29     return str(next(ID_GENERATOR))
30 
31 
32 def parse_inventory(path):
33     inventory = InventoryManager(loader=DataLoader(), sources=path)
34     return inventory.get_groups_dict()
35 
36 
37 def attach_machine_id(result, mid):
38     key = "machine_id"
39     if isinstance(result, list):
40         for r in result:
41             r[key] = mid
42     else:
43         result[key] = mid
44     return result
45 
46 
47 def process_archives(archives):
48     for archive in archives:
49         with extract(archive) as ex:
50             ctx = create_context(ex.tmp_dir)
51             broker = dr.Broker()
52             broker[ctx.__class__] = ctx
53             yield dr.run(broker=broker)
54 
55 
56 def extract_facts(brokers):
57     results = defaultdict(list)
58     for b in brokers:
59         mid = b[machine_id]
60         for k, v in b.get_by_type(plugins.fact).items():
61             r = attach_machine_id(v, mid)
62             if isinstance(r, list):
63                 results[k].extend(r)
64             else:
65                 results[k].append(r)
66     return results
67 
68 
69 def process_facts(facts, meta, broker):
70     broker[ClusterMeta] = meta
71     for k, v in facts.items():
72         broker[k] = pd.DataFrame(v)
73     return dr.run(dr.COMPONENTS[dr.GROUPS.cluster], broker=broker)
74 
75 
76 def process_cluster(archives, broker, inventory=None):
77     inventory = parse_inventory(inventory) if inventory else {}
78 
79     brokers = process_archives(archives)
80     facts = extract_facts(brokers)
81     meta = ClusterMeta(len(archives), inventory)
82 
83     return process_facts(facts, meta, broker)
84
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/insights/core/cluster.py b/insights/core/cluster.py
--- a/insights/core/cluster.py
+++ b/insights/core/cluster.py
@@ -1,8 +1,10 @@
 #!/usr/bin/env python
 import itertools
-import pandas as pd
+import os
 from collections import defaultdict
 
+import pandas as pd
+
 from ansible.parsing.dataloader import DataLoader
 from ansible.inventory.manager import InventoryManager
 
@@ -46,8 +48,14 @@
 
 def process_archives(archives):
     for archive in archives:
-        with extract(archive) as ex:
-            ctx = create_context(ex.tmp_dir)
+        if os.path.isfile(archive):
+            with extract(archive) as ex:
+                ctx = create_context(ex.tmp_dir)
+                broker = dr.Broker()
+                broker[ctx.__class__] = ctx
+                yield dr.run(broker=broker)
+        else:
+            ctx = create_context(archive)
             broker = dr.Broker()
             broker[ctx.__class__] = ctx
             yield dr.run(broker=broker)
diff --git a/insights/core/hydration.py b/insights/core/hydration.py
--- a/insights/core/hydration.py
+++ b/insights/core/hydration.py
@@ -37,13 +37,15 @@
     if not common_path:
         raise archives.InvalidArchive("Unable to determine common path")
 
-    if any(f.endswith(archives.COMPRESSION_TYPES) for f in os.listdir(common_path)):
-        return common_path, ClusterArchiveContext
-
     return common_path, HostArchiveContext
 
 
 def create_context(path, context=None):
+    top = os.listdir(path)
+    arc = [os.path.join(path, f) for f in top if f.endswith(archives.COMPRESSION_TYPES)]
+    if arc:
+        return ClusterArchiveContext(path, all_files=arc)
+
     all_files = get_all_files(path)
     if not all_files:
         raise archives.InvalidArchive("No files in archive")

{"golden_diff": "diff --git a/insights/core/cluster.py b/insights/core/cluster.py\n--- a/insights/core/cluster.py\n+++ b/insights/core/cluster.py\n@@ -1,8 +1,10 @@\n #!/usr/bin/env python\n import itertools\n-import pandas as pd\n+import os\n from collections import defaultdict\n \n+import pandas as pd\n+\n from ansible.parsing.dataloader import DataLoader\n from ansible.inventory.manager import InventoryManager\n \n@@ -46,8 +48,14 @@\n \n def process_archives(archives):\n     for archive in archives:\n-        with extract(archive) as ex:\n-            ctx = create_context(ex.tmp_dir)\n+        if os.path.isfile(archive):\n+            with extract(archive) as ex:\n+                ctx = create_context(ex.tmp_dir)\n+                broker = dr.Broker()\n+                broker[ctx.__class__] = ctx\n+                yield dr.run(broker=broker)\n+        else:\n+            ctx = create_context(archive)\n             broker = dr.Broker()\n             broker[ctx.__class__] = ctx\n             yield dr.run(broker=broker)\ndiff --git a/insights/core/hydration.py b/insights/core/hydration.py\n--- a/insights/core/hydration.py\n+++ b/insights/core/hydration.py\n@@ -37,13 +37,15 @@\n     if not common_path:\n         raise archives.InvalidArchive(\"Unable to determine common path\")\n \n-    if any(f.endswith(archives.COMPRESSION_TYPES) for f in os.listdir(common_path)):\n-        return common_path, ClusterArchiveContext\n-\n     return common_path, HostArchiveContext\n \n \n def create_context(path, context=None):\n+    top = os.listdir(path)\n+    arc = [os.path.join(path, f) for f in top if f.endswith(archives.COMPRESSION_TYPES)]\n+    if arc:\n+        return ClusterArchiveContext(path, all_files=arc)\n+\n     all_files = get_all_files(path)\n     if not all_files:\n         raise archives.InvalidArchive(\"No files in archive\")\n", "issue": "Directories of exploded archives not recognized as cluster\nDirectories of exploded archives aren't recognized as a cluster but are erroneously identified as whatever context matches the first marker file we encounter after recursively enumerating every file in all subdirectories.\n", "before_files": [{"content": "import logging\nimport os\nfrom itertools import product\n\nfrom insights.core import archives\nfrom insights.core.context import (ClusterArchiveContext,\n                                   JDRContext,\n                                   HostArchiveContext,\n                                   SosArchiveContext,\n                                   SerializedArchiveContext)\n\nlog = logging.getLogger(__name__)\n\n\ndef get_all_files(path):\n    all_files = []\n    for f in archives.get_all_files(path):\n        if os.path.isfile(f) and not os.path.islink(f):\n            all_files.append(f)\n    return all_files\n\n\ndef identify(files):\n    markers = {\"insights_archive.txt\": SerializedArchiveContext,\n               \"insights_commands\": HostArchiveContext,\n               \"sos_commands\": SosArchiveContext,\n               \"JBOSS_HOME\": JDRContext}\n\n    for f, m in product(files, markers):\n        if m in f:\n            i = f.find(m)\n            common_path = os.path.dirname(f[:i])\n            ctx = markers[m]\n            return common_path, ctx\n\n    common_path = os.path.dirname(os.path.commonprefix(files))\n    if not common_path:\n        raise archives.InvalidArchive(\"Unable to determine common path\")\n\n    if any(f.endswith(archives.COMPRESSION_TYPES) for f in os.listdir(common_path)):\n        return common_path, ClusterArchiveContext\n\n    return common_path, HostArchiveContext\n\n\ndef create_context(path, context=None):\n    all_files = get_all_files(path)\n    if not all_files:\n        raise archives.InvalidArchive(\"No files in archive\")\n\n    common_path, ctx = identify(all_files)\n    context = context or ctx\n    return context(common_path, all_files=all_files)\n", "path": "insights/core/hydration.py"}, {"content": "#!/usr/bin/env python\nimport itertools\nimport pandas as pd\nfrom collections import defaultdict\n\nfrom ansible.parsing.dataloader import DataLoader\nfrom ansible.inventory.manager import InventoryManager\n\nfrom insights.core import dr, plugins\nfrom insights.core.archives import extract\nfrom insights.core.hydration import create_context\nfrom insights.specs import Specs\n\n\nID_GENERATOR = itertools.count()\n\n\nclass ClusterMeta(dict):\n    def __init__(self, num_members, kwargs):\n        self.num_members = num_members\n        self.update(**kwargs)\n\n\n@plugins.combiner(optional=[Specs.machine_id, Specs.hostname])\ndef machine_id(mid, hn):\n    ds = mid or hn\n    if ds:\n        return ds.content[0].strip()\n    return str(next(ID_GENERATOR))\n\n\ndef parse_inventory(path):\n    inventory = InventoryManager(loader=DataLoader(), sources=path)\n    return inventory.get_groups_dict()\n\n\ndef attach_machine_id(result, mid):\n    key = \"machine_id\"\n    if isinstance(result, list):\n        for r in result:\n            r[key] = mid\n    else:\n        result[key] = mid\n    return result\n\n\ndef process_archives(archives):\n    for archive in archives:\n        with extract(archive) as ex:\n            ctx = create_context(ex.tmp_dir)\n            broker = dr.Broker()\n            broker[ctx.__class__] = ctx\n            yield dr.run(broker=broker)\n\n\ndef extract_facts(brokers):\n    results = defaultdict(list)\n    for b in brokers:\n        mid = b[machine_id]\n        for k, v in b.get_by_type(plugins.fact).items():\n            r = attach_machine_id(v, mid)\n            if isinstance(r, list):\n                results[k].extend(r)\n            else:\n                results[k].append(r)\n    return results\n\n\ndef process_facts(facts, meta, broker):\n    broker[ClusterMeta] = meta\n    for k, v in facts.items():\n        broker[k] = pd.DataFrame(v)\n    return dr.run(dr.COMPONENTS[dr.GROUPS.cluster], broker=broker)\n\n\ndef process_cluster(archives, broker, inventory=None):\n    inventory = parse_inventory(inventory) if inventory else {}\n\n    brokers = process_archives(archives)\n    facts = extract_facts(brokers)\n    meta = ClusterMeta(len(archives), inventory)\n\n    return process_facts(facts, meta, broker)\n", "path": "insights/core/cluster.py"}], "after_files": [{"content": "import logging\nimport os\nfrom itertools import product\n\nfrom insights.core import archives\nfrom insights.core.context import (ClusterArchiveContext,\n                                   JDRContext,\n                                   HostArchiveContext,\n                                   SosArchiveContext,\n                                   SerializedArchiveContext)\n\nlog = logging.getLogger(__name__)\n\n\ndef get_all_files(path):\n    all_files = []\n    for f in archives.get_all_files(path):\n        if os.path.isfile(f) and not os.path.islink(f):\n            all_files.append(f)\n    return all_files\n\n\ndef identify(files):\n    markers = {\"insights_archive.txt\": SerializedArchiveContext,\n               \"insights_commands\": HostArchiveContext,\n               \"sos_commands\": SosArchiveContext,\n               \"JBOSS_HOME\": JDRContext}\n\n    for f, m in product(files, markers):\n        if m in f:\n            i = f.find(m)\n            common_path = os.path.dirname(f[:i])\n            ctx = markers[m]\n            return common_path, ctx\n\n    common_path = os.path.dirname(os.path.commonprefix(files))\n    if not common_path:\n        raise archives.InvalidArchive(\"Unable to determine common path\")\n\n    return common_path, HostArchiveContext\n\n\ndef create_context(path, context=None):\n    top = os.listdir(path)\n    arc = [os.path.join(path, f) for f in top if f.endswith(archives.COMPRESSION_TYPES)]\n    if arc:\n        return ClusterArchiveContext(path, all_files=arc)\n\n    all_files = get_all_files(path)\n    if not all_files:\n        raise archives.InvalidArchive(\"No files in archive\")\n\n    common_path, ctx = identify(all_files)\n    context = context or ctx\n    return context(common_path, all_files=all_files)\n", "path": "insights/core/hydration.py"}, {"content": "#!/usr/bin/env python\nimport itertools\nimport os\nfrom collections import defaultdict\n\nimport pandas as pd\n\nfrom ansible.parsing.dataloader import DataLoader\nfrom ansible.inventory.manager import InventoryManager\n\nfrom insights.core import dr, plugins\nfrom insights.core.archives import extract\nfrom insights.core.hydration import create_context\nfrom insights.specs import Specs\n\n\nID_GENERATOR = itertools.count()\n\n\nclass ClusterMeta(dict):\n    def __init__(self, num_members, kwargs):\n        self.num_members = num_members\n        self.update(**kwargs)\n\n\n@plugins.combiner(optional=[Specs.machine_id, Specs.hostname])\ndef machine_id(mid, hn):\n    ds = mid or hn\n    if ds:\n        return ds.content[0].strip()\n    return str(next(ID_GENERATOR))\n\n\ndef parse_inventory(path):\n    inventory = InventoryManager(loader=DataLoader(), sources=path)\n    return inventory.get_groups_dict()\n\n\ndef attach_machine_id(result, mid):\n    key = \"machine_id\"\n    if isinstance(result, list):\n        for r in result:\n            r[key] = mid\n    else:\n        result[key] = mid\n    return result\n\n\ndef process_archives(archives):\n    for archive in archives:\n        if os.path.isfile(archive):\n            with extract(archive) as ex:\n                ctx = create_context(ex.tmp_dir)\n                broker = dr.Broker()\n                broker[ctx.__class__] = ctx\n                yield dr.run(broker=broker)\n        else:\n            ctx = create_context(archive)\n            broker = dr.Broker()\n            broker[ctx.__class__] = ctx\n            yield dr.run(broker=broker)\n\n\ndef extract_facts(brokers):\n    results = defaultdict(list)\n    for b in brokers:\n        mid = b[machine_id]\n        for k, v in b.get_by_type(plugins.fact).items():\n            r = attach_machine_id(v, mid)\n            if isinstance(r, list):\n                results[k].extend(r)\n            else:\n                results[k].append(r)\n    return results\n\n\ndef process_facts(facts, meta, broker):\n    broker[ClusterMeta] = meta\n    for k, v in facts.items():\n        broker[k] = pd.DataFrame(v)\n    return dr.run(dr.COMPONENTS[dr.GROUPS.cluster], broker=broker)\n\n\ndef process_cluster(archives, broker, inventory=None):\n    inventory = parse_inventory(inventory) if inventory else {}\n\n    brokers = process_archives(archives)\n    facts = extract_facts(brokers)\n    meta = ClusterMeta(len(archives), inventory)\n\n    return process_facts(facts, meta, broker)\n", "path": "insights/core/cluster.py"}]}